/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.NTable;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.client.RowFilter;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.LeaseHolder;
import com.nhncorp.neptune.common.LeaseListener;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.NeptuneLock;
import com.nhncorp.neptune.common.TimeLog;
import com.nhncorp.neptune.common.aop.ProxyFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.TestCaseTabletServerKillException;
import com.nhncorp.neptune.common.io.NWritableUtils;
import com.nhncorp.neptune.common.ipc.AclManager;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.common.ipc.NRPC.Server;
import com.nhncorp.neptune.common.util.Daemon;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.neptune.common.util.SizeOf;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.master.NeptuneMaster;
import com.nhncorp.neptune.master.TabletMasterProtocol;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.RowColumnValues;
import com.nhncorp.neptune.tablet.TableSchema;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tablet.TabletReport;
import com.nhncorp.neptune.tabletserver.action.BatchUploadAction;
import com.nhncorp.neptune.tabletserver.action.TabletAction;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventFilter;
import com.nhncorp.pleiades.client.LockEventMonitor;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.protocol.LockMetaData;

/**
 * Tablet을 관리하는 메인 서버<br>
 * Neptune을 구성하는 세가지 데몬(NeptuneMaster, TabletServer, ChangeLogServer) 서버 중 하나이다.<p> 
 * 다음과 같은 기능을 수행한다.<br>
 * <li>1. Tablet의 데이터 서비스(get, put)</li> 
 * 
 * @author nhn
 *
 */
public class TabletServer implements TabletServerIF {
  //FIXME TabletDrop 처리 시 기존에 처리 중이던 작업에 대해서는 어떻게 하나???
  public static final Log LOG = LogFactory.getLog(TabletServer.class.getName());

  private static final int MAX_TABLET_LOAD_THREAD = 10;
  
  private static final int MAX_ACTION_THREAD = 50;
  
  private int maxTabletCount = 400;

  protected NConfiguration conf;

  private String hostName;

  private Server server;

  private NeptuneFileSystem fs;

  // tabletName, tablet
  private Map<String, Tablet> tablets = Collections.synchronizedMap(new HashMap<String, Tablet>());

  //private Set<TabletInfo> stopedTablets = Collections.synchronizedSet(new HashSet<TabletInfo>(100));
  
  // TxId, tabletName
  private Map<TxId, String> tabletTxs = Collections.synchronizedMap(new HashMap<TxId, String>());

  // scannerId, tabletName
  private Map<String, String> tabletScanners = Collections.synchronizedMap(new HashMap<String, String>(10));

  private boolean stopRequested = false;

  private TabletMasterProtocol masterServer;
  
  protected LeaseHolder leaseHolder;

  //protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  protected final NeptuneLock lock = new NeptuneLock();

  // tableName, Table
  protected Map<String, TableSchema> tableSchemas = Collections.synchronizedMap(new HashMap<String, TableSchema>(10));

  protected boolean daemonMode = false;   //TestCase를 위해 설정
  
  //Scan, Bulk upload 등과 같이 많은 데이터 처리시 Stream 처리를 위해 사용
  private Daemon dataXceiveServer;
  
  //private HeartbeatThread heartbeatThread;
  private MinorCompactionThread minorCompactionThread;
  
  private MajorCompactionThread majorCompactionThread;
  //private TabletSplitThread tabletSplitThread;
  
  private TabletServerIF proxyObject;
  
  private Count scanCount = new Count(0);
  
  private Count uploaderCount = new Count(0);
  
  //protected TabletActionListener tabletActionListener;
  
  private int txTimeout;
  
  private String tabletServerLockPath;

  protected ThreadGroup threadGroup;
  
  private boolean testMode;
  
  private ExecutorService loadThreadExecutor = Executors.newFixedThreadPool(MAX_TABLET_LOAD_THREAD);
  
  protected ThreadPoolExecutor compactionExecutor;
  
  protected ThreadPoolExecutor splitExecutor;
  
  protected ThreadPoolExecutor actionExecutor = (ThreadPoolExecutor)Executors.newFixedThreadPool(MAX_ACTION_THREAD);

  protected  AsyncTaskManager asyncTaskManager = new AsyncTaskManager(); 
  
  protected Date serverStartTime;
  
  static private ThreadMXBean threadBean = 
    ManagementFactory.getThreadMXBean();
  
  //TableName -> TargetCount, DropedCount
  //protected Map<String, Integer[]> dropingTablets = new HashMap<String, Integer[]>();
  
  //TabletName -> BatchUploadAction
  private Map<String, Map<String, BatchUploadAction>> batchUploadActions = 
    Collections.synchronizedMap(new HashMap<String, Map<String, BatchUploadAction>>(10));
  
  //TabletServer에 메모리 부하가 발생한 경우 apply 처리 후 delay 시킨다.
  private AtomicBoolean heavyMemory = new AtomicBoolean(false);
  private AtomicBoolean heavyThread = new AtomicBoolean(false);
  
  private TabletServerTxStatistics txStatistics;
  
  private int heartbeatPort;
  
  private LockService lockService;
  
  private int maxMajorCompactionThread;
  
  private int maxSplitThread;
  
  private Random applyRandom = new Random();
  
  public TabletServer()  {
  }
  
  public void init(TabletServerIF proxyObject, NConfiguration conf) throws IOException {
    this.serverStartTime = new Date();
    this.testMode = conf.getBoolean("testmode", false);
    this.proxyObject = proxyObject;
    this.conf = conf;

    maxMajorCompactionThread = this.conf.getInt("tabletServer.maxMajorCompactionThread", 5);
    maxSplitThread = this.conf.getInt("tabletServer.maxSplitThread", 10);
    
    compactionExecutor = (ThreadPoolExecutor)Executors.newFixedThreadPool(maxMajorCompactionThread);
    
    splitExecutor = (ThreadPoolExecutor)Executors.newFixedThreadPool(maxSplitThread);
    
    //ROOT, META tabletserver로 처리를 요청하는 경우 짧은 시간에 한번만 retry한다.(필요하면 해당 로직에서 retry)
//    this.metaConf = new NConfiguration(conf);
//    this.metaConf.set("ipc.client.timeout", 5*1000);
//    this.metaConf.set("ipc.client.connect.max.retries", 5);

    this.maxTabletCount = conf.getInt("tabletServer.max.tablet.count", 300);
    
    this.fs = NeptuneFileSystem.get(conf);

    InetSocketAddress serverAddress = NetworkUtil.getAddress(InetAddress.getLocalHost().getHostName() + ":"
        + conf.getInt("tabletServer.port", 7001));
    
    this.hostName = serverAddress.getHostName() + ":" + serverAddress.getPort();
    
    this.threadGroup = new ThreadGroup("TabletServer_" + hostName);
    
    this.leaseHolder = new LeaseHolder(threadGroup);
    
    this.tabletServerLockPath = Constants.SERVER + "/" + NetworkUtil.hostNameToFileName(hostName);
    
    this.heartbeatPort = conf.getInt("tabletServer.heartbeat.port", 50300);
    //create session for heartbeat
    this.lockService = LockServiceFactory.getLockService(conf, hostName, true, heartbeatPort);
    
    //<Split Lock 삭제>
    lockService.deleteNode(getSplitLockPath(), true);
    //</Split Lock 삭제>
    
    this.server = NRPC.getServer(lockService, proxyObject, serverAddress.getHostName(), serverAddress.getPort(), 
            conf.getInt("tabletServer.handler.count", 10), false, conf);
    
    ServerSocket ss = null;
    int port = conf.getInt("tabletServer.scanner.port", 50100);
    String bindAddress = "0.0.0.0";
    
    try {
//      LOG.info("Opened Scanner Handler at " + hostName  + ", port=" + port);
      ss = new ServerSocket(port,0,InetAddress.getByName(bindAddress));
      ss.setReuseAddress(true);
    } catch (IOException ie) {
      LOG.error("Could not open scanner server at " + port + ", stop server and Stop tablet server", ie);
      exit();
    }
    this.dataXceiveServer = new Daemon(new DataXceiveServer(ss));  
    //this.tabletActionListener = new TabletServerActionListener();
    
    this.txTimeout = conf.getInt("client.tx.timeout", 60) * 1000;
    
    if(!acquireServerLock(true)) {
      LOG.fatal("TabletServer stopped. Can't server lock:" + tabletServerLockPath);
      exit();
    }
    
    addLockEventHandler();
    
    txStatistics = new TabletServerTxStatistics(conf, threadGroup);
//    LOG.debug("TabletServer [" + this + "," + hostName + "] init");
    
//    Properties props = System.getProperties();
//    LOG.debug("=================================================================");
//    for(Map.Entry<Object, Object> entry: props.entrySet()) {
//      LOG.debug(entry.getKey() + ":" + entry.getValue());
//    }
//    LOG.debug("=================================================================");
  }

  /**
   * TabletServer에서 처리해야 하는 lock event에 대한 call back 메소드를 등록한다.
   * @throws IOException
   */
  private void addLockEventHandler() throws IOException {
    LockEventMonitor sessionExipreEventMonitor = new LockEventMonitor() {

      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, tabletServerLockPath);
        
        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.SESSION_EXPIRED);
        lockEvents.add(LockService.Events.NODE_REMOVED);
        filter.setLockEventList(lockEvents);
        
        return filter;
      }

      //Session이 expire되는 경우 네트워크 문제 등으로 인해 Pleiades에 lock을 유지 못하는 경우이다.
      //이 경우 TabletServer는 lock 파일이 존재하면 lock을 획득하는 시도를 하고
      //lock 파일이 없는 경우 자동으로 종료한다.
      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        while(true) {
          //lock 파일이 존재하지 않으면 종료
          try {
            boolean lockStatus = checkTabletServerLock();
            if(lockStatus) {
              break;
            }
            try {
              Thread.sleep(5*1000);
            } catch (InterruptedException e) {
              return;
            }
          } catch (Exception e) {
            LOG.error(e);
          }
        }
      }
    };
    
    //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
    if(lockService.getEventHandler() != null) {
      lockService.getEventHandler().addLockEventMonitor(sessionExipreEventMonitor);
    }
    
    //DropTableMonitor dropTableEventMonitor = new DropTableMonitor(proxyObject);
    //lockService.getEventHandler().addLockEventMonitor(dropTableEventMonitor);
  }
  
  private boolean checkTabletServerLock() {
    //LockService lockService = null;
    try {
      //lockService = LockServiceFactory.getLockService(conf, hostName, true);
      if(lockService == null) {
        lockService = LockServiceFactory.getLockService(conf, hostName, true);
      }
      
      if(lockService == null) {
        LOG.fatal("TabletServer stopped. cause lockService is null");
        shutdown();
        return false;
      }
      
      //자신의 lock 정보가 존재하지 않으면 shutdown
      if(!lockService.exist(tabletServerLockPath)) {
        //Thread.dumpStack();
        LOG.fatal("TabletServer stopped. lock not exist:" + tabletServerLockPath);
        shutdown();
        return false;
      } else {
        //lock이 존재하는 경우에도 tryLock에서 false가 반환되면 shutdown
        if(!lockService.tryLock(tabletServerLockPath, LockService.LockMode.LOCK_EXCLUSIVE)) {
          LOG.fatal("TabletServer stopped. can't get lock:" + tabletServerLockPath);
          shutdown();
          return false;
        } else {
          return true;
        }
      }
    } catch (IOException e) {
      //Thread.dumpStack();
      LOG.fatal("TabletServer stopped. session expired:" + tabletServerLockPath + "," + e.getMessage(), e);
      shutdown();
      return false;
    }
  }
  
  private boolean acquireServerLock(boolean overwrite) throws IOException {
    LockMetaData lockMetaData = new LockMetaData(tabletServerLockPath);
    lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
    //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
    if(!overwrite && lockService.getMetaData(tabletServerLockPath) != null) {
      //LOG.debug("Get TabletServer Lock:" + tabletServerLockPath);
      return lockService.getLock(tabletServerLockPath, LockService.LockMode.LOCK_EXCLUSIVE);
    } else {
      //LOG.debug("Get TabletServer Lock:" + tabletServerLockPath);
      return lockService.createNode(lockMetaData, overwrite);
    }
  }
  
  public String getTabletServerConf(String key) throws IOException {
    return conf.get(key);
  }
  
  public String getHostName() {
    return this.hostName;
  }
  
  public void printTabletInfo(String tabletName) throws IOException {
    Tablet tablet = getTablet(tabletName);
    System.out.println("=====================================");
    System.out.println("          " + tabletName);
    System.out.println("-------------------------------------");
    if (tablet == null) {
      System.out.println("No tablet");
    } else {
      tablet.print();
    }
    System.out.println("=====================================");
  }

  public String[] getAllActions(String tabletName) throws IOException {
    Tablet tablet = getTablet(tabletName);

    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }
    
    return tablet.getAllActions();
  }
  
  public void assignTablet(TabletInfo tabletInfo) throws IOException {
    LOG.debug("assigned tablet: on " + hostName + ", tablet="+ tabletInfo);
    lock.obtainReadLock();
    try {
      if(tablets.size() >= maxTabletCount) {
        LOG.warn("Can't load table[" +  tabletInfo.getTabletName() + "] cause by tablets.size()[" + 
            tablets.size() + "] exceeds MAX_TABLET_COUNT[" + maxTabletCount + "]");
        connectMasterServer().errorTabletAssignment(hostName, tabletInfo);
      }
    } catch (IOException e) {
      LOG.error("Can't load table[" +  tabletInfo.getTabletName() + "]", e);
      return;
    } finally {
      lock.releaseReadLock();
    }
    
    loadThreadExecutor.execute(new TabletLoadTask(this, tabletInfo));
    //Thread tabletLoadThread = new Thread(threadGroup, new TabletLoadThread(this, tabletInfo));
    //tabletLoadThread.start();
  }

  public TabletInfo[] reportTablets() {
    lock.obtainReadLock();
    try {
      int index = 0;
      //LOG.debug("ReportTablets: " + this + "," + hostName + ": " + tablets.size());
      TabletInfo[] tabletInfos = new TabletInfo[tablets.size()];
      for (Tablet eachTablet : tablets.values()) {
        tabletInfos[index++] = eachTablet.getTabletInfo();
        //LOG.debug("ReportTablets: " + hostName + ": " + eachTablet.getTabletInfo().getTabletName());
      }
      return tabletInfos;
    } finally {
      lock.releaseReadLock();
    }
  }

  
  public void insertRow(String tabletName, Row.Key rowKey, String[] columnNames, ColumnValue[][] columnValues) throws IOException {
    if(columnNames == null || columnNames.length == 0 || columnValues == null || columnValues.length == 0) {
      return;
    }
    
    if(columnValues.length != columnNames.length) {
      throw new IOException("not equals column's length and value's length");
    }
    TxId txId = null;
    
    long startTime = System.currentTimeMillis();
    int retry = 0;
    
    while(true) {
      txId = startTx(tabletName, rowKey);
      if(txId != null) {
        break;
      }
      try {
        Thread.sleep(50);
      } catch (InterruptedException e1) {
      }        
      retry++;
  
      if( (System.currentTimeMillis() - startTime) > txTimeout) {
        break;
      }
    }
    
    if(txId == null) {
      throw new IOException("fails after " + txTimeout + " ms. rowKey=" + rowKey + ", tabletName=" + tabletName);
    }
    try {
      insertColumnValue(txId, columnNames, columnValues);
      commit(txId);
    } catch (IOException e) {
      rollback(txId);
      throw e;
    }
  }
  
  public void insertColumnValue(TxId txId, String columnName, ColumnValue[] columnValues) throws IOException {
    leaseHolder.touch(txId);

    Tablet tablet = checkTxId(txId);
    tablet.insertColumnValue(txId, columnName, columnValues);
  }
  
  public void insertColumnValue(TxId txId, String[] columnNames, ColumnValue[][] columnValues) throws IOException {
    leaseHolder.touch(txId);
    if(columnNames == null || columnNames.length == 0) {
      return;
    }
    Tablet tablet = checkTxId(txId);
    for(int i = 0; i < columnNames.length; i++) {
      if(columnValues[i] != null && columnValues[i].length > 0) {
        tablet.insertColumnValue(txId, columnNames[i], columnValues[i]);
      }
    }
    leaseHolder.touch(txId);
  }

  public void commit(TxId txId) throws IOException {
    commit(txId, true);
  }

  public void commit(TxId txId, boolean saveLog) throws IOException {
    Tablet tablet = checkTxId(txId);
    tablet.commit(txId, saveLog);
    try {
      lock.obtainWriteLock();
      try {
        tabletTxs.remove(txId);
        //LOG.debug("Remove\t" + txId);
      } finally {
        lock.releaseWriteLock();
      }
      leaseHolder.removeLease(txId);
    } catch (Exception e) {
      LOG.error("while removing txId" + e.getMessage());
    }
  }
  
  public void commit(TxId txId, ChangeLog[] changeLogList, boolean saveLog) throws IOException {
    Tablet tablet = checkTxId(txId);
    fillTimestamp(changeLogList);
    
    tablet.commit(txId, changeLogList, saveLog);
    try {
      lock.obtainWriteLock();
      try {
        tabletTxs.remove(txId);
      //  LOG.debug("Remove\t" + txId);
      } finally {
        lock.releaseWriteLock();
      }
      leaseHolder.removeLease(txId);
    } catch (Exception e) {
      LOG.error("while removing txId" + e.getMessage());
    }
  }
  
  private void fillTimestamp(ChangeLog[] changeLogList) {
    // by sangchul
    // timestamp로써 역할을 할 수 있을까? 값이 다 똑같진 않을까?
    for(ChangeLog log : changeLogList) {
      if (log.timestamp == ChangeLog.USE_SERVER_TIMESTAMP) {
        log.timestamp = System.currentTimeMillis();
      }
    }
  }

  public boolean apply(String tabletName, Row.Key rowKey, TxId txId, ChangeLog[] changeLogList, boolean saveLog) throws IOException {
    fillTimestamp(changeLogList);
    try { 
      long startTime = System.nanoTime();
      boolean memory = heavyMemory.get();
      boolean thread = heavyThread.get();
      
      if(thread && compactionExecutor.getActiveCount() >= maxSplitThread * (0.9)) {
        if(applyRandom.nextInt() % 4 == 0) {
          txStatistics.addTx(TabletServerTxStatistics.TYPE_PUT_SKIP, 0);
          return false;
        }
      }
      
      boolean success = getTabletByName(tabletName).apply(rowKey, txId, changeLogList, saveLog);
      
      long time = 0;
      
//      if(memory) {
//        time = memoryLoadSleep;
//      } else if(thread) {
//        if(compactionExecutor.getActiveCount() >= MAX_MAJOR_COMPACTION_THREAD * (0.9)) {
//          time = peakLoadSleep;
//        } else {
//          time = avgLoadSleep;
//        }
//      }
//      
//      if(time > 0) {
//        try {
//          Thread.sleep(time);
//        } catch (InterruptedException e) {
//          return false;
//        }
//      }
      if(success) {
        txStatistics.addTx(TabletServerTxStatistics.TYPE_PUT, (System.nanoTime() - startTime));
      } else {
        txStatistics.addTx(TabletServerTxStatistics.TYPE_PUT_FAIL, (System.nanoTime() - startTime));
      }
      return success;
    } catch (IOException e) {
      IOException err = new IOException(rowKey + "," + e.getMessage());
      err.initCause(e);
      throw err;
    }
  }
  
  public TxId startTx(String tabletName, Row.Key rowKey) throws IOException {
    return startTx(tabletName, rowKey, true);
  }
  
  public TxId startTx(String tabletName, Row.Key rowKey, boolean systemTimestamp)
      throws IOException {
    TxId txId = getTabletByName(tabletName).startTx(rowKey, systemTimestamp);

    if (txId == null) {
      // by sangchul
      // 이 경우 뭔가 예외를 발생해야 할 듯...
      return null;
    }

    lock.obtainWriteLock();
    try {
      tabletTxs.put(txId, tabletName);
      // LOG.debug("add\t" + txId);
    } finally {
      lock.releaseWriteLock();
    }

    // int txExpireTime = conf.getInt("client.tx.timeout", 30) * 1000;
    leaseHolder.createLease(txId, new TxListener(tabletName, txId), txTimeout);

    return txId;
  }

  private Tablet getTabletByName(String tabletName) throws IOException {
    Tablet ret = null;
    lock.obtainReadLock();
    try {
      ret = tablets.get(tabletName);
    } finally {
      lock.releaseReadLock();
    }

    if(ret == null) {
      throw new IOException("no tablet[" + tabletName + "] in tablet server@" 
          + InetAddress.getLocalHost().getHostName());
    }
    
    return ret;
  }

  public Row.Key rollback(TxId txId) throws IOException {
    Row.Key rowKey = null;
    try {
      Tablet tablet = checkTxId(txId);
      rowKey = tablet.rollback(txId);
    } catch (IOException e) {
      
    }
    
    lock.obtainWriteLock();
    try {
      tabletTxs.remove(txId);
//      LOG.info(">>>>>>>>>>Remove(Rollback):" + txId);
//      Thread.dumpStack();
    } finally {
      lock.releaseWriteLock();
    }
    leaseHolder.removeLease(txId);

    return rowKey;
  }

  public ColumnValue get(String tabletName, Row.Key rowKey, 
      String columnName, Cell.Key columnKey) throws IOException {
    TimeLog.checkPoint();
    Tablet tablet = getTablet(tabletName);

    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }

    AclManager.checkPermission(
        conf,
        lockService,
        tableSchemas, tablet.getTabletInfo().getTableName(), "r");
    
    ColumnValue result = tablet.get(rowKey, columnName, columnKey);
    TimeLog.endCheckPoint("TabletServer: get1: " + rowKey);
    return result;
  }

  public RowColumnValues[][] gets(String tabletName, Row.Key startRowKey,
      RowFilter rowFilter) throws IOException {
    Tablet tablet = getTablet(tabletName);
    
    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }
    
    rowFilter.setStartRowKey(startRowKey);
    String[] columnNames = rowFilter.getColumns();
    
    TabletScanner[] scanners = new TabletScanner[columnNames.length];
     
    for(int i = 0; i < columnNames.length; i++) {
      scanners[i] = tablet.openScanner(columnNames[i], rowFilter);
      if(scanners[i]  == null) {
        throw new IOException("Can't open scanner:" + tabletName);
      }
      String scannerId = scanners[i].getScannerId();
      lock.obtainWriteLock();
      try {
        tabletScanners.put(scannerId, tabletName);
      } finally {
        lock.releaseWriteLock();
      }      
      int scannerExpireTime = conf.getInt("client.tx.timeout", 30) * 2 * 1000;
      leaseHolder.createLease(scannerId, new TabletScannerListener(tabletName, scannerId), scannerExpireTime);
    }
    
    List<RowColumnValues[]> result = new ArrayList<RowColumnValues[]>();
    ServerSideMultiColumnScanner multiColumnScanner = null;
    try {
      multiColumnScanner = new ServerSideMultiColumnScanner(this, rowFilter, scanners);
  
      RowColumnValues[] rowColumnValue = null;
      while((rowColumnValue = multiColumnScanner.nextRow()) != null) {
        result.add(rowColumnValue);
        if(multiColumnScanner.getNumScannedColumnValue() >= NTable.MAX_MULTI_ROW_RESULT) {
          break;
        }
      }
      return result.toArray(new RowColumnValues[][]{});
    } finally {
      if(multiColumnScanner != null) {
        multiColumnScanner.close();
      }
      for(int i = 0; i < scanners.length; i++) {
        if(scanners[i] != null) {
          try {
            closeScanner(scanners[i].getScannerId());
          } catch (Exception e) {
            LOG.error("Can't close scanner:" + e.getMessage(), e);
          }
        }
      }
    }
  }
  
  public RowColumnValues[] get(String tabletName, RowFilter scanFilter) throws IOException {
    //TimeLog.checkPoint();
    long startTime = System.nanoTime();
    Tablet tablet = getTablet(tabletName);
  
    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }
    AclManager.checkPermission(
        conf,
        lockService,
        tableSchemas, tablet.getTabletInfo().getTableName(), "r");
    
    RowColumnValues[] result = tablet.get(scanFilter);
    txStatistics.addTx(TabletServerTxStatistics.TYPE_GET, (System.nanoTime() - startTime));
    //TimeLog.endCheckPoint("TabletServer: get3:" + scanFilter.getRowKey());
    return result;
  }
  
  public ColumnValue[][] get(String tabletName, Row.Key rowKey) throws IOException {
    TimeLog.checkPoint();
    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }
    TableSchema table = descTable(tablet.getTabletInfo().getTableName());
    
    AclManager.checkPermission(
        conf,
        lockService,
        tableSchemas, tablet.getTabletInfo().getTableName(), "r");
    
    List<String> columnList = table.getColumns();
    ColumnValue[][] result = get(tabletName, rowKey, columnList.toArray(new String[columnList.size()]));
    TimeLog.endCheckPoint("TabletServer: get4: " + rowKey);
    return result;
  }


  public ColumnValue[][] get(String tabletName, Row.Key rowKey, String[] columnNames) throws IOException {
    //long startTime = System.currentTimeMillis();
    Tablet tablet = getTablet(tabletName);
    if (tablet == null || !tablet.isReady()) {
      throw new IOException("no tablet[" + tabletName + "] on " + hostName);
    }

    AclManager.checkPermission(
        conf,
        lockService,
        tableSchemas, tablet.getTabletInfo().getTableName(), "r");
    
    ColumnValue[][] result =  tablet.get(rowKey, columnNames);
//    long endTime = System.currentTimeMillis();
//    synchronized(this) {
//      getTimeSum += (endTime - startTime);
//      getCount++;
//      if(getCount % 1000 == 0) {
//        LOG.info("getTimeSum:" + getTimeSum);
//        getTimeSum = 0;
//      }
//    }
    return result;
  }
  
  public Tablet getTablet(String tabletName) {
    Tablet tablet = null;
    lock.obtainReadLock();
    try {
      tablet = tablets.get(tabletName);
    } finally {
      lock.releaseReadLock();
    }
    return tablet;
  }

  private TableSchema descTable(String tableName) throws IOException {
    lock.obtainWriteLock();
    try {
      if (tableSchemas.containsKey(tableName)) {
        return tableSchemas.get(tableName);
      } else {
        TableSchema table = TableSchema.loadTableSchema(conf, lockService, tableName);
        tableSchemas.put(tableName, table);

        return table;
      }
    } finally {
      lock.releaseWriteLock();
    }
  }

  private ColumnValue[] next(String scannerId) throws IOException {
    String tabletName = null;
    lock.obtainReadLock();
    try {
      tabletName = tabletScanners.get(scannerId);
    } finally {
      lock.releaseReadLock();
    }

    if (tabletName == null) {
      throw new IOException("[" + scannerId + "] not opened");
    }
    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
     throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
    }
    leaseHolder.touch(scannerId);

    return tablet.scanNext(scannerId);
  }
  
  public void touch(String leaseId) throws IOException {
    leaseHolder.touch(leaseId);
  }

  /**
   * 
   */
  private String openScanner(String tabletName, String columnName, RowFilter scanFilter)
      throws IOException {
    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      throw new IOException("[" + tabletName + "] not serviced on this server [" + hostName + "]");
    }

    // FIXME Scanner가 오픈되어있을때 Minor, Major Compaction이 발생하면 어떻게 하는가?
    TabletScanner tabletScanner = tablet.openScanner(columnName, scanFilter);
    if(tabletScanner == null) {
      return null;
    }
    String scannerId = tabletScanner.getScannerId();

    lock.obtainWriteLock();
    try {
      tabletScanners.put(scannerId, tabletName);
    } finally {
      lock.releaseWriteLock();
    }

    int scannerExpireTime = conf.getInt("client.tx.timeout", 30) * 2 * 1000;
    leaseHolder.createLease(scannerId, new TabletScannerListener(tabletName, scannerId), scannerExpireTime);
    return scannerId;
  }

  protected void closeScanner(String scannerId) throws IOException {
    String tabletName = null;
    lock.obtainReadLock();
    try {
      tabletName = tabletScanners.get(scannerId);
    } finally {
      lock.releaseReadLock();
    }
    if (tabletName == null) {
      throw new IOException("[" + scannerId + "] not opened");
    }
    
    Tablet tablet = getTablet(tabletName);
    if (tablet != null) {
      tablet.closeScanner(scannerId);
    }
    lock.obtainWriteLock();
    try {
      tabletScanners.remove(scannerId);
    } finally {
      lock.releaseWriteLock();
    }
    leaseHolder.removeLease(scannerId);
  }

  public void removeRow(TxId txId, long timestamp) throws IOException {
    leaseHolder.touch(txId);

    Tablet tablet = checkTxId(txId);
    tablet.removeRow(txId, timestamp);
  }

  public void run() {
    try {
      daemonMode = true;
//      markRunningTabletServer();  modified by Terry

      this.server.start();

      minorCompactionThread = new MinorCompactionThread();
      minorCompactionThread.start();
      majorCompactionThread = new MajorCompactionThread();
      majorCompactionThread.start();
      
      Thread tabletServerLockCheckerThread = new Thread(threadGroup, new TabletServerLockChecker());
      tabletServerLockCheckerThread.start();
      
      dataXceiveServer.start();
    } catch (IOException e) {
      e.printStackTrace();
      LOG.error(e.getMessage(), e);
      exit();
    }

    LOG.info("TabletServer[" + hostName + "] started");
    try {
      this.server.join();
    } catch (InterruptedException ie) {
      return;
    }
  }

  /**
   * 
   */
  public void deleteColumnValue(TxId txId, String columnName, Cell.Key columnKey) throws IOException {
    leaseHolder.touch(txId);
    deleteColumnValue(txId, columnName, columnKey, System.currentTimeMillis());
  }

  /**
   * 
   * @param txId
   * @param columnName
   * @param columnKey
   * @param timestamp
   * @throws IOException
   */
  private boolean deleteColumnValue(TxId txId, String columnName, Cell.Key columnKey, long timestamp) throws IOException {
    leaseHolder.touch(txId);
    Tablet tablet = checkTxId(txId);
    return tablet.deleteColumnValue(txId, columnName, columnKey, timestamp);
  }

  public boolean existsTxId(TxId txId) throws IOException {
    try {
      Tablet tablet = checkTxId(txId);
      if(tablet != null) {
        return true;
      } else {
        return false;
      }
    } catch (Exception e) {
      return false;
    }
  }
  
  /**
   * 
   * @param txId
   * @return
   * @throws IOException
   */
  private Tablet checkTxId(TxId txId) throws IOException {
    if(txId == null || txId.getTxId() == null || txId.getTxId().equals("null")) {
      throw new IOException("TxId is null");
    }
    String tabletName = null;
    lock.obtainReadLock();
    try {
      tabletName = tabletTxs.get(txId);
    } finally {
      lock.releaseReadLock();
    }

    if (tabletName == null) {
      throw new IOException("TxId[" + txId + "] not exits");
    }

    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
    }
    return tablet;
  }

  /**
   * 
   */
  public void updateColumnValue(TxId txId, String columnName, ColumnValue[] records) throws IOException {
    updateColumnValue(txId, columnName, records, false);
  }
  
  
  public void updateOrInsertColumnValue(TxId txId, String columnName, ColumnValue[] records) throws IOException {
    updateColumnValue(txId, columnName, records, true);
  }
  
  /**
   * 
   */
  private void updateColumnValue(TxId txId, String columnName, ColumnValue[] records, boolean ignoreNoData) throws IOException {
    leaseHolder.touch(txId);

    for (ColumnValue columnValue : records) {
      long timestamp = txId.isSystemTimestamp() ? System.currentTimeMillis() : columnValue.getTimestamp();

      // delete
      if(!deleteColumnValue(txId, columnName, columnValue.getCellKey(), timestamp - 1)) {
        if(!ignoreNoData) {
          throw new IOException("updateColumnValue: " + columnName + ", rowkey=" + columnValue.getRowKey() + ", columnKey=" + columnValue.getCellKey() + " record not exist");
        }
      }

      // insert
      insertColumnValue(txId, columnName, new ColumnValue[] { columnValue });
    }
  }

  public boolean dropTable(final String taskId, final String tableName) throws IOException {
    //META에서 Tablet을 삭제하고 TabletServer의 변수에서 삭제한다.
    //데이터 파일에 대한 삭제는 NeptuneMaster에서 일괄 삭제한다.
    LOG.debug("Drop table:" + hostName + ": " + tableName);
    lock.obtainWriteLock();
    final List<Tablet> dropTablets = new ArrayList<Tablet>();
    try {
      for(Map.Entry<String, Tablet> entry: tablets.entrySet()) {
        Tablet tablet = entry.getValue();
        if(tableName.equals(tablet.getTabletInfo().getTableName())) {
          dropTablets.add(tablet);
        }
      }
      tableSchemas.remove(tableName);
    } finally {
      lock.releaseWriteLock();
    }
    
    if(dropTablets.size() == 0) {
      return false;
    } 
    
    Thread dropThread = new Thread(threadGroup, new Runnable() {
      public void run() {
        try {
          SizeOf.runGC();
          LOG.info("Memory Status before drop: free=" + SizeOf.freeMemory() + ", total=" + SizeOf.totalMemory());
          NTable ntable = NTable.openTable(conf, TABLE_NAME_META);
          for(Tablet eachTablet: dropTablets) {
            TabletInfo tabletInfo = eachTablet.getTabletInfo();
            
            lock.obtainWriteLock();
            try {
              tablets.remove(tabletInfo.getTabletName());
            } finally {
              lock.releaseWriteLock();
            } 

            //META에서 Tablet 정보를 삭제한다.
            LOG.debug("Drop table:delete from META:" + tabletInfo);
            Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
            ntable.remove(rowKey, Constants.META_COLUMN_NAME_TABLETINFO, new Cell.Key(tabletInfo.getTabletName()));

            eachTablet.drop();
            
            eachTablet = null;
          }
          connectMasterServer().endTableDrop(taskId, hostName, tableName);
        } catch (Exception e) {
          LOG.error("Drop error:" + tableName, e);
          try {
            connectMasterServer().errorTableDrop(taskId, hostName, tableName, e.getMessage());
          } catch (IOException e1) {
            LOG.fatal("Error while report error table drop:" + tableName, e1);
          }
        } finally {
          try {
            SizeOf.runGC();
          } catch (IOException e) {
            e.printStackTrace();
          }
          LOG.info("Memory Status after drop: free=" + SizeOf.freeMemory() + ", total=" + SizeOf.totalMemory());
        }
      }
    });

    dropThread.start();
    
    return true;
  }
  
  /**
   * Root tablet에 META tablet에 대한 정보를 등록한다.
   * 
   * @param tabletInfo
   * @throws IOException
   */
  private void insertIntoRootTable(TabletInfo tabletInfo, long timestamp) throws IOException {
    if (!TABLE_NAME_META.equals(tabletInfo.getTableName())) {
      throw new IOException("tablet must be MetaTable");
    }
    int retry = 0;
    DataServiceProtocol rootTabletServer = null;
    TxId txId = null;
    Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
    ColumnValue columnValue = new ColumnValue(rowKey, new Cell.Key(tabletInfo.getTabletName()), tabletInfo.getWriteBytes(), timestamp);
    
    while(retry < 30) {
      try {      
        TabletInfo rootTabletInfo = Tablet.getRootTabletInfo(conf, lockService);//LockServiceFactory.getLockService(conf, hostName, true));
        
        rootTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
            DataServiceProtocol.versionID, NetworkUtil.getAddress(rootTabletInfo.getAssignedHostName()), conf);

        txId = rootTabletServer.startTx(rootTabletInfo.getTabletName(), rowKey);
        if(txId == null) {
          LOG.error("startTx recrived null txId");
          retry++;
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e1) {
            return;
          }
          continue;
        }
        
        rootTabletServer.insertColumnValue(txId, META_COLUMN_NAME_TABLETINFO, new ColumnValue[] { columnValue });
        
        rootTabletServer.commit(txId);
        break;
      } catch (Exception e) {
        LOG.error("insertIntoRootTable", e);
        //e.printStackTrace();
        try {
          rootTabletServer.rollback(txId);
        } catch (Exception e1) {
        }
        
        retry++;
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e1) {
          return;
        }
      }
    }
    if(retry >= 30) {
      LOG.error("fail insertIntoRootTable to ROOT after 30 times retry");
      throw new IOException("fail insertIntoRootTable to ROOT after 30 times retry");
    }     
  }

  /**
   * META Table에 TabletInfo에 대한 정보를 저장한다.
   * 
   * @param tabletInfo
   * @throws IOException
   */
  protected String insertIntoMetaTable(TabletInfo tabletInfo, long timestamp) throws IOException {
    //LOG.info("Insert into META tablet:" + hostName + ":" + tabletInfo);
    Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
    int retry = 0;
    DataServiceProtocol metaTabletServer = null;
    ColumnValue columnValue = new ColumnValue(rowKey, 
                                              new Cell.Key(tabletInfo.getTabletName()), 
                                              tabletInfo.getWriteBytes(),
                                              timestamp);
    TxId txId = null;
    StringBuffer tabletNameBuffer = null;
    Exception err = null;
    while(retry < 30) {
      try {        
        tabletNameBuffer = new StringBuffer();
        metaTabletServer = connectMetaTabletServer(tabletInfo, tabletNameBuffer, rowKey);
        if(metaTabletServer == null) {
          retry++;
          Thread.sleep(1000);
          continue;
        }
        txId = metaTabletServer.startTx(tabletNameBuffer.toString(), rowKey);
        if(txId == null) {
          retry++;
          Thread.sleep(1000);
          continue;
        }
        
        metaTabletServer.insertColumnValue(txId, META_COLUMN_NAME_TABLETINFO, new ColumnValue[] { columnValue });
    
        metaTabletServer.commit(txId);
    
        //LOG.info("Inserted into META tablet:" + hostName + ":" + tabletInfo + ",MetaServer" + tabletNameBuffer.toString());
        return tabletNameBuffer.toString();
      } catch (Exception e) {
        LOG.error("Exception in insertIntoMetaTable tabletName : " + tabletNameBuffer.toString() + ", rowKey : " + rowKey, e);
        err = e;
        try {
          metaTabletServer.rollback(txId);
        } catch (Exception e1) {
          LOG.error("Exception in rollback tabletName : " + tabletNameBuffer.toString() + ", rowKey : " + rowKey, e1);
        }
        
        retry++;
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e1) {
          return null;
        }
      }
    }
    if(retry >= 30) {
      LOG.error("fail insertIntoMetaTable to " + tabletNameBuffer + " after 30 times retry:" + err);
      throw new IOException("fail insertIntoMetaTable to " + tabletNameBuffer + " after 30 times retry");
    }
    return null;
  }

  /**
   * 
   * @param tabletInfo
   * @throws IOException
   */
  protected void modifyRootTable(TabletInfo tabletInfo) throws IOException {
    if (!TABLE_NAME_META.equals(tabletInfo.getTableName())) {
      throw new IOException("tablet must be MetaTable");
    }
    //ROOT가 재할당 중인 경우에는 Root Tablet에 대한 info가 부정확할 수 있다. 따라서 retry 한다.
    long startTime = System.currentTimeMillis();
    TabletInfo rootTabletInfo = null;
    DataServiceProtocol rootTabletServer = null;
    while(true) {
      try {
        rootTabletInfo = Tablet.getRootTabletInfo(conf, lockService);//LockServiceFactory.getLockService(conf, hostName, true));
        rootTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
            DataServiceProtocol.versionID, NetworkUtil.getAddress(rootTabletInfo.getAssignedHostName()), conf);
        break;
      } catch (Exception e) {
      }
    }
    
    Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());

    int retry = 0;
    while(retry < 30) {
      TxId txId = rootTabletServer.startTx(rootTabletInfo.getTabletName(), rowKey);
      if(txId == null) {
        retry++;
        continue;
      }
      try {
        ColumnValue columnValue = new ColumnValue(rowKey, new Cell.Key(tabletInfo.getTabletName()), tabletInfo.getWriteBytes());
        rootTabletServer.updateOrInsertColumnValue(txId, META_COLUMN_NAME_TABLETINFO, new ColumnValue[] { columnValue });
  
        rootTabletServer.commit(txId);
        break;
      } catch (Exception e) {
        //LOG.error(e);
        try {
          rootTabletServer.rollback(txId);
        } catch (Exception e1) {
        }
        
        retry++;
        
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e1) {
          return;
        }
      }
    }
    if(retry >= 30) {
      throw new IOException("fail modifyRootTable after 30 times retry");
    }      
    //System.out.println("Root update time>>>>>" + (System.currentTimeMillis() - startTime));
  }

  /**
   * 
   * @param tabletInfo
   * @throws IOException
   */
  protected void modifyMetaTable(TabletInfo tabletInfo) throws IOException {
    StringBuffer metaTabletNameBuffer = new StringBuffer(10);
    Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
    int retry = 0;
    DataServiceProtocol metaTabletServer = null;
    TxId txId = null;
    while(retry < 10) {
      try {    
        metaTabletNameBuffer = new StringBuffer(10);
        metaTabletServer = connectMetaTabletServer(tabletInfo, metaTabletNameBuffer, rowKey);
        txId = metaTabletServer.startTx(metaTabletNameBuffer.toString(), rowKey);
        if(txId == null) {
          retry++;
          continue;
        }
        ColumnValue columnValue = new ColumnValue(rowKey, new Cell.Key(tabletInfo.getTabletName()), tabletInfo.getWriteBytes());
        metaTabletServer.updateOrInsertColumnValue(txId, META_COLUMN_NAME_TABLETINFO, new ColumnValue[] { columnValue });
    
        metaTabletServer.commit(txId);
        //metaTabletServer.printMeta(metaTabletNameBuffer.toString());
        break;
      } catch (Exception e) {
        if(retry == 0) {
          LOG.warn("Can't modifyMetaTable cause " + e.getMessage() + ", byt retry:" + tabletInfo, e);
        } else {
          LOG.warn("Can't modifyMetaTable cause " + e.getMessage() + ", byt retry:" + tabletInfo);
        }
        try {
          metaTabletServer.rollback(txId);
        } catch (Exception e1) {
        }
        
        retry++;
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e1) {
          return;
        }
      }
    }
    if(retry >= 10) {
      LOG.error("fail modifyMetaTable to " + metaTabletNameBuffer + " after 10 times retry");
      throw new IOException("fail modifyMetaTable to " + metaTabletNameBuffer + " after 10 times retry");
    }        
  }

  /**
   * 
   * @param tabletInfo
   * @throws IOException
   */
//  public static void deleteFromRootTable(NeptuneConfiguration conf, 
//      TabletInfo tabletInfo, 
//      long timestamp) throws IOException {
//    if (!TABLE_NAME_META.equals(tabletInfo.getTableName())) {
//      throw new IOException("tablet must be MetaTable");
//    }
//    Row.Key rowKey = Tablet.generateMetaRow.Key(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
//    NTable ntable = NTable.openTable(conf, TABLE_NAME_ROOT);
//    if(ntable == null) {
//      return;
//    }
//    ntable.startTx(rowKey);
//    ntable.removeRow(timestamp);
//    ntable.commit();
//  }

  /**
   * 
   * @param tabletInfo
   * @throws IOException
   */
//  public static boolean deletFromMetaTable(NeptuneConfiguration conf,
//      Row.Key rowKey, 
//      long timestamp) throws IOException {
//    //Row.Key rowKey = Tablet.generateMetaRow.Key(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
//    NTable ntable = NTable.openTable(conf, TABLE_NAME_META);
//    if(ntable == null) {
//      LOG.error("Can't open META Table");
//      return false;
//    }
//    ntable.startTx(rowKey);
//    ntable.removeRow(timestamp);
//    ntable.commit();
//    
//    //LOG.info("Delete TabletInfo from META:" + rowKey);
//    return true;
//  }
  
  private DataServiceProtocol connectMetaTabletServer(TabletInfo tabletInfo, StringBuffer metaTabletNameBuffer,
                                                        Row.Key rowKey) throws IOException {
    TabletInfo rootTabletInfo = Tablet.getRootTabletInfo(conf, lockService);//LockServiceFactory.getLockService(conf, hostName, true));
    DataServiceProtocol rootTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
        DataServiceProtocol.versionID, NetworkUtil.getAddress(rootTabletInfo.getAssignedHostName()), conf);

    // find meta tablet from root tablet
    Row.Key rootTabletSearchRowKey = Tablet.generateMetaRowKey(TABLE_NAME_META, rowKey);
    ColumnValue metaTabletRecord = rootTabletServer.getCloestMetaData(rootTabletInfo.getTabletName(), 
                                      rootTabletSearchRowKey);
    if(metaTabletRecord == null) {
      //LOG.error("meta tablet info is null[" + tabletInfo + ", rowKey=" + rowKey + "]");
      return null;
    }
    TabletInfo metaTablet = new TabletInfo();
    metaTablet.readFields(metaTabletRecord.getValue());
    
    String metatabletName = metaTablet.getTabletName();
    String assignedHost = metaTablet.getAssignedHostName();
    
    if (assignedHost == null) {
      // FIXME meta tablet이 할당되지 않았을 경우에 대한 처리
      LOG.error("meta tablet [" + metatabletName + "] not assigned");
      return null;
    }
    DataServiceProtocol metaTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
        DataServiceProtocol.versionID, NetworkUtil.getAddress(assignedHost), conf);

    metaTabletNameBuffer.append(metatabletName);
    return metaTabletServer;
  }

  /**
   * 
   * @param targetTablet
   * @param splitedTablets
   * @throws IOException
   */
  public void tabletSplited(Tablet targetTablet, Tablet[] splitedTablets) throws IOException {
    //Tablet 클래스를 테스트하는 TestCase에서 수행한 경우 
    if(!daemonMode) {
      return;
    }
    
    if(!tablets.containsKey(targetTablet.getTabletInfo().getTabletName()) ||
        targetTablet.isDroped()) {
      //Drop 처리된 경우 split은 무효화 된다.
      throw new IOException("Can't split cause droped tablet:" + targetTablet.getTabletInfo().getTabletName());
    }
    
    //////////////////////////////////////////////////////////////////
    //META 수정 후 TabletServer가 kill 되는 상황에 대비해서 split 된 정보 생성
    //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
    if(lockService == null) {
      return;
    }
    
    String splitLockPath = getSplitLockPath(targetTablet);
    
    LockMetaData lockMetaData = new LockMetaData(splitLockPath);
    //tableName + maxRow.Key, maxRow.Key
    //구분자가 rowkey내에 있는 경우 문제 발생, 향후 pleiades에 byte를 저장할 수 있도록 수정
    DecimalFormat df = new DecimalFormat("00000");
    String contentKey = df.format(targetTablet.getTabletInfo().getTableName().getBytes().length) + 
                        targetTablet.getTabletInfo().getTableName() +
                        df.format(splitedTablets[0].getTabletInfo().getTabletName().getBytes().length) + 
                        splitedTablets[0].getTabletInfo().getTabletName() +  
                        df.format(splitedTablets[0].getTabletInfo().getEndRowKey().getBytes().length) + 
                        splitedTablets[0].getTabletInfo().getEndRowKey() + 
                        df.format(splitedTablets[1].getTabletInfo().getTabletName().getBytes().length) +
                        splitedTablets[1].getTabletInfo().getTabletName() +
                        df.format(splitedTablets[1].getTabletInfo().getEndRowKey().getBytes().length) +
                        splitedTablets[1].getTabletInfo().getEndRowKey();
    
    lockMetaData.setContentKey(contentKey);
    lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
    lockService.createNode(lockMetaData, true);
    //////////////////////////////////////////////////////////////////
    
    //Split 되었지만 finish 작업 도중 오류가 발생한 경우 META 데이터를 
    //이용하여 불필요한 tablet에 대한 내용을  파일시스템에서 삭제하는 로직이 필요. --> NeptuneMaster에서 수행 
    proxyObject.modifyMetaDataForSplit(targetTablet, splitedTablets);

    lock.obtainWriteLock();
    try {
      //두번째 Tablet은 자신이 서비스하고 첫번째 Tablet은 다른 TabletServer로 할당시킨다.
      tablets.remove(targetTablet.getTabletInfo().getTabletName());
      tablets.put(splitedTablets[1].getTabletInfo().getTabletName(), splitedTablets[1]);
    } catch (Exception e) {
      //장애 발생시 자동 종료(장애 발생 확율은 0% 이지만 그래도 혹시나 발생시는 자신을 스스로 kill
      LOG.fatal("TableServer shuwdowned cause fail to remove/put after split", e);
      shutdown();
    } finally {
      lock.releaseWriteLock();
    }

    proxyObject.reportTabletSplited(targetTablet, splitedTablets);
    
    lockService.deleteNode(splitLockPath);
    lockService.deleteNode(Constants.UPLOADER + "/" + targetTablet.getTabletInfo().getTabletName(), true);
  }

  private String getSplitLockPath(Tablet targetTablet) {
    return Constants.TABLETSERVER_SPLIT + "/" + NetworkUtil.hostNameToFileName(hostName) + "/" + targetTablet.getTabletInfo().getTabletName();
  }
  
  private String getSplitLockPath() {
    return Constants.TABLETSERVER_SPLIT + "/" + NetworkUtil.hostNameToFileName(hostName);
  }

  public void reportTabletSplited(Tablet targetTablet, Tablet[] splitedTablets) throws IOException {
    try {
      TabletInfo[] splitedTabletInfos = new TabletInfo[] { splitedTablets[0].getTabletInfo(),
          splitedTablets[1].getTabletInfo() };
  
      //여기서 장애가 발생하는 경우는 NeptuneMaster가 정상적인 상황인 아닌 경우
      //따라서 NeptuneMaster가 재 시작되면 Split된 Tablet은 다시 다른 TabletServer로 할당됨
      //TODO drop 처리 중인 상황에 대한 동기화 처리
      connectMasterServer().reportTabletSplited(targetTablet.getTabletInfo(), splitedTabletInfos);
    } catch (TestCaseTabletServerKillException e) {
      throw e;
    } catch (Exception e) {
      LOG.info("Tablet creation report fail", e);
    }
  }

  public void modifyMetaDataForSplit(Tablet targetTablet, Tablet[] splitedTablets) throws IOException {
    if(targetTablet.isDroped()) {
      return;
    }
    /*
     * META의 경우 하나의 Table에 대한 정보는 하나의 META tablet에서 관리하도록 split한다.
     * (이 부분은 아직 구현되지 않았음)
     * 따라서 여기서 하나의 tablet이 split 되었을 경우 발생하는 세개의 tx(delete old, insert new1, insert new2)를
     * Tablet에서는 하나의 tx 처리하는 것과 동일하게 처리한다.
     */    
    int maxRetry = 50;
    try {
      //처리 도중 kill 되는 경우에 META에는 변경된 내용이 반영 안됨
      if (TABLE_NAME_META.equals(targetTablet.getTabletInfo().getTableName())) {
        //META table인 경우
        int retry = 0;
        while(retry < maxRetry) {
          try {
            TabletInfo rootTabletInfo = Tablet.getRootTabletInfo(conf, lockService); //LockServiceFactory.getLockService(conf, hostName, true));
            DataServiceProtocol rootTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
                DataServiceProtocol.versionID, NetworkUtil.getAddress(rootTabletInfo.getAssignedHostName()), conf);    
            
            rootTabletServer.saveTabletSplitedInfo(
                                    rootTabletInfo.getTabletName(), 
                                    targetTablet.getTabletInfo(), 
                                    new TabletInfo[]{splitedTablets[0].getTabletInfo(), 
                                      splitedTablets[1].getTabletInfo()});
            break;
          } catch (Exception e) {
            retry++;
            try {
              Thread.sleep(1000);
            } catch (InterruptedException e1) {
                return;
            }
          }
        }
        if(retry >= maxRetry) {
          throw new IOException("fail saveTabletSplitedInfo to ROOT");
        }
      } else {
        //일반 table인 경우
        Row.Key rowKey = Tablet.generateMetaRowKey(targetTablet.getTabletInfo().getTableName(), targetTablet.getTabletInfo().getEndRowKey());
        StringBuffer metaTabletName = new StringBuffer(10);
        int retry = 0;
        Exception err = null;
        while(retry < maxRetry) {
          try {
            metaTabletName = new StringBuffer(10);
            DataServiceProtocol metaTabletServer = this.connectMetaTabletServer(targetTablet.getTabletInfo(), metaTabletName, rowKey);
            metaTabletServer.saveTabletSplitedInfo(metaTabletName.toString(), targetTablet.getTabletInfo(), 
                                                                new TabletInfo[]{splitedTablets[0].getTabletInfo(), splitedTablets[1].getTabletInfo()});
            break;
          } catch (Exception e) {
            retry++;
            err = e;
            try {
              Thread.sleep(1000);
            } catch (InterruptedException e1) {
              return;
            }
          }
        }
        if(retry >= maxRetry) {
          IOException exception = new IOException("fail saveTabletSplitedInfo to " + metaTabletName + "," + err);
          exception.initCause(err);
          throw exception;
        }
      }
    } catch (Exception e) {
      IOException exception = new IOException(e.getMessage());
      exception.initCause(e);
      throw exception;
    }
  }

  public void saveTabletSplitedInfo(String tabletName, TabletInfo targetTablet, TabletInfo[] splitedTablets) throws IOException {
    //LOG.debug("saveTabletSplitedInfo:" + tabletName + ", targetTablet=" + targetTablet);
    final Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      LOG.error("saveTabletSplitedInfo: [" + tabletName + "] not serviced on this server(" + hostName + ")");
      throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
    }
    
    String tableName = tablet.getTabletInfo().getTableName();
    if(!TABLE_NAME_ROOT.equals(tableName) && !TABLE_NAME_META.equals(tableName)) {
      LOG.error("saveTabletSplitedInfo: [" + tableName + "] is not META or ROOT");
      throw new IOException(" [" + tableName + "] is not META or ROOT");
    }
    
    Row.Key rowKey = Tablet.generateMetaRowKey(targetTablet.getTableName(), targetTablet.getEndRowKey());
    TxId txId = tablet.startTx(rowKey);
    if(txId == null) {
      throw new IOException("Tablet is busy. Can't start Tx [" + rowKey + "]");
    }
    try {
      tablet.saveTabletSplitedInfo(txId, targetTablet, splitedTablets);
      tablet.commit(txId, true);
    } catch (Exception e) {
      LOG.error("Error in saving tablet splited info. tabletName [" + tabletName 
          + ", txID [" + txId + "], targetTablet [" + targetTablet.getTabletName() + "]", e);
      tablet.rollback(txId);
    }
  }
  
  public void stopAllTablets() throws IOException {
    tablets.clear();
  }
  
  public static void main(String args[]) throws IOException {
    try {
      Thread.sleep(10 * 1000);
    } catch (InterruptedException e) {
      return;
    }    
    try {
      TabletServerIF tabletServer = (TabletServerIF)ProxyFactory.createProxyClass("tabletServer");
      tabletServer.init(tabletServer, new NConfiguration());
      (new Thread(tabletServer)).start();
    } catch (Exception e) {
      LOG.fatal("Error while start TabletServer:" + e.getMessage(), e);
      System.exit(0);
    }
  }

  private TabletMasterProtocol connectMasterServer() throws IOException {
    return connectMasterServer(3);
  }
  
  private TabletMasterProtocol connectMasterServer(int maxRetry) throws IOException {
    int retry = 0;
    while (retry < maxRetry) {
      try {
        if (masterServer == null) {
          String masterHostName = NeptuneMaster.getMasterServerHostName(lockService);//LockServiceFactory.getLockService(conf, hostName, true));
          masterServer = (TabletMasterProtocol) NRPC.getProxy(TabletMasterProtocol.class,
                TabletMasterProtocol.versionID, NetworkUtil.getAddress(masterHostName), conf);
        }
        masterServer.checkServer();
        break;
      } catch (Exception e) {
        masterServer = null;
        retry++;
        if (retry >= maxRetry) {
          break;
        }
        try {
          Thread.sleep(100);
        } catch (InterruptedException e1) {
          return null;
        }
      }
    }
    return masterServer;
  }
  
  public TabletInfo getTabletInfo(String tabletName) {
    Tablet tablet = getTablet(tabletName);
    if (tablet == null)
      return null;

    return tablet.getTabletInfo();
  }

  public boolean isServicedTablet(String tabletName) {
    return tablets.containsKey(tabletName);
  }

  public ColumnValue getCloestMetaData(String tabletName, Row.Key rowKey) throws IOException {
    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      LOG.debug("getCloest:[" + tabletName + "] not serviced on this server(" + hostName + ")");
      return null;
    }

    //LOG.debug("getNearest:" + tabletName + ":" + rowKey);
    return tablet.getClosestMeta(rowKey);
  }

  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
    if (protocol.equals(DataServiceProtocol.class.getName())) {
      return DataServiceProtocol.versionID;
    } else if (protocol.equals(TabletManagerProtocol.class.getName())) {
      return TabletManagerProtocol.versionID;
    } else {
      throw new IOException("Unknown protocol to tablet server: " + protocol);
    }
  }

  public boolean checkServer() {
    return true;
  }

  public void splitForTest(TabletInfo tabletInfo) throws IOException {
    Tablet tablet = getTablet(tabletInfo.getTabletName());
    if (tablet == null) {
      LOG.debug("splitForTest:[" + tabletInfo.getTabletName() + "] not serviced on this server(" + hostName + ")");
    }
    if (tablet.isReady()) {
      tablet.doTabletSplit();
    }
  }
  
  /**
   * 테스트 코드를 위해 NeptuneMaster로 전송하는 부분을 인터페이스로 빼서 처리.
   */
  public void endTabletAssignmentForHandler(TabletInfo tabletInfo, boolean create) throws IOException {
    connectMasterServer().endTabletAssignment(tabletInfo, create);
  }
  
  public void stopAction(String tableName, String tabletActionClassName) throws IOException {
    lock.obtainReadLock();
    try {
      LOG.info("stopAction:" + tableName + "," + tabletActionClassName);
      for(Tablet eachTablet: tablets.values()) {
        TabletInfo tabletInfo = eachTablet.getTabletInfo();
        if(tabletInfo.getTableName().equals(tableName)) {
          TabletAction tabletAction = (TabletAction)Class.forName(tabletActionClassName).newInstance();
          eachTablet.actionChecker.endActionType(tabletAction);
          
          if(BatchUploadAction.class.getName().equals(tabletActionClassName)) {
            //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
            lockService.deleteNode(Constants.UPLOADER + "/" + tabletInfo.getTabletName(), true);
            batchUploadActions.remove(tabletInfo.getTabletName());
          }
        }
      }
    } catch (Exception e) {
      throw new IOException("endAction:" + tabletActionClassName + ":" + e.getMessage());
    } finally {
      lock.releaseReadLock();
    }
  }
  
  private void exit() {
    try {
      final int STACK_DEPTH = 20;
      boolean contention = threadBean.isThreadContentionMonitoringEnabled();
      long[] threadIds = threadBean.getAllThreadIds();
      LOG.info("Process Thread Dump");
      LOG.info(threadIds.length + " active threads");
      for (long tid: threadIds) {
        ThreadInfo info = threadBean.getThreadInfo(tid, STACK_DEPTH);
        if (info == null) {
          LOG.info("  Inactive");
          continue;
        }
        LOG.info("Thread " + info.getLockOwnerId() + "," + info.getLockOwnerName());
        Thread.State state = info.getThreadState();
        LOG.info("  State: " + state);
        LOG.info("  Blocked count: " + info.getBlockedCount());
        LOG.info("  Waited count: " + info.getWaitedCount());
        if (contention) {
          LOG.info("  Blocked time: " + info.getBlockedTime());
          LOG.info("  Waited time: " + info.getWaitedTime());
        }
        if (state == Thread.State.WAITING) {
          LOG.info("  Waiting on " + info.getLockName());
        } else  if (state == Thread.State.BLOCKED) {
          LOG.info("  Blocked on " + info.getLockName());
          LOG.info("  Blocked by " + info.getLockOwnerId() + "," + info.getLockOwnerName());
        }
        LOG.info("  Stack:");
        for (StackTraceElement frame: info.getStackTrace()) {
          LOG.info("    " + frame.toString());
        }
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
    
    System.exit(0);
  }
  
  public void shutdown() {
    LOG.fatal("Kill Tablet Server:" + hostName + ",freeMemory:" + SizeOf.freeMemory() + "/" + SizeOf.totalMemory());
    if(!testMode) {
      try {
        System.out.println("Kill Tablet Server:" + hostName + ",freeMemory:" + SizeOf.freeMemory() + "/" + SizeOf.totalMemory());
      } catch (Exception e) {
        LOG.error(e);
      }
      exit();
    } else {
    //if (!stopRequested) {
      stopRequested = true;
      tablets.clear();
      
      try {
        server.stop();
      } catch (Exception e) {
        LOG.error(e);
      }
      try {
        ((DataXceiveServer) this.dataXceiveServer.getRunnable()).kill();
      } catch (Exception e) {
        LOG.error(e);
      }
      
      try {
        if(minorCompactionThread != null) {
          minorCompactionThread.interrupt();
        }
      } catch (Exception e) {
        LOG.error(e);
      }
      try {
        if(majorCompactionThread != null) {
          majorCompactionThread.interrupt();
        }
      } catch (Exception e) {
        LOG.error(e);
      }
      
      try {
        //ThreadGroup threadGroup = Thread.currentThread().getThreadGroup();
        Thread[] threads = new Thread[threadGroup.activeCount()];
        threadGroup.enumerate(threads, true);
        
        for(Thread thread: threads) {
          try {
            thread.interrupt();
          } catch (Exception e) {
            
          }
        }       
      } catch (Exception e) {
        LOG.error(e);
      }

      try {
        loadThreadExecutor.shutdownNow();
        compactionExecutor.shutdownNow();
        splitExecutor.shutdownNow();
      } catch (Exception e) {
        LOG.error(e);
      }

      try {
        //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
        if(lockService != null) {
          lockService.close();
          //lockService.poison(true);
        }
      } catch (Exception e) {
        LOG.error(e);
      }
      
      try {
        LockServiceFactory.close(this.hostName);
      } catch (Exception e) {
        LOG.error(e);
      }
      LOG.debug("shutdown tabletserver:" + hostName);
    }
  }
  
  public String startBatchUploader(String tabletName) throws IOException {
    //FIXME 동일한 Tablet에 대해 이미 batch upload가 수행중인 경우???
    Tablet tablet = getTablet(tabletName);
    if(tablet == null) {
      throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
    }
    
    //runBatchUpload 중인 경우에는 split/merge가 안되게
    //split, merge 수행중인 경우에는 close(), 클라이언트에서 다시 처리하도록
    BatchUploadAction action = new BatchUploadAction();
    
    synchronized(batchUploadActions) {
      //FIXME lease 처리하여 일정시간동안 upload가 없으면 중지 처리
      boolean result =  tablet.actionChecker.startAction(action);
      if(result) {
        Map<String, BatchUploadAction> actions = batchUploadActions.get(tabletName);
        if(actions == null) {
          actions = new HashMap<String, BatchUploadAction>();
          batchUploadActions.put(tabletName, actions);
        }
        //TabletServer가 kill 되었을 경우 새로 할당받은 TabletServer에서도 계속 처리 가능하도록 lock 설정
        LockMetaData lockMetaData = new LockMetaData(Constants.UPLOADER + "/" + tabletName + "/" + action.getActionKey());
        lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
        //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
        if(!lockService.createNode(lockMetaData, true)) {
          return null;
        }
        actions.put(action.getActionKey(), action);
        
        return action.getActionKey();
      } else {
        return null;
      }
    }
  }

  public void cancelBatchUploader(String actionId, String tabletName) throws IOException {
    LOG.debug("cancelBatchUploader:" + tabletName + "," + actionId);
    Tablet tablet = getTablet(tabletName);
    if(tablet == null ) {
      LOG.warn("[" + tabletName + "] not serviced on this server(" + hostName + ")");
      return;
    }
    
    Map<String, BatchUploadAction> actions = null;
    
    synchronized(batchUploadActions) {
      if(!batchUploadActions.containsKey(tabletName)) {
        LOG.warn("No opened Uploader: " + tabletName);
        return;
      }
      
      actions = batchUploadActions.get(tabletName);
      
      if(!actions.containsKey(actionId)) {
        LOG.warn("No opened Uploader: " + tabletName + "," + actionId + ",host=" + hostName);
        return;
      }

      lockService.deleteNode(Constants.UPLOADER + "/" + tabletName + "/" + actionId, true);
      tablet.actionChecker.endAction(actions.get(actionId));
      actions.remove(actionId);
    }
  }
  
  public String endBatchUploader(
      String actionId,
      String tabletName, 
      String[] columnNames, 
      String[] mapFileIds,
      String[] mapFilePaths) throws IOException {
    Tablet tablet = getTablet(tabletName);
    if(tablet == null ) {
      LOG.warn("[" + tabletName + "] not serviced on this server(" + hostName + ")");
      return null;
    }
    synchronized(batchUploadActions) {
      if(!batchUploadActions.containsKey(tabletName)) {
        LOG.warn("No opened Uploader: " + tabletName);
        return null;
      }
      
      if(!batchUploadActions.get(tabletName).containsKey(actionId)) {
        LOG.warn("No opened Uploader: " + tabletName + "," + actionId + ",host=" + hostName);
        return null;
      }
    }
    String taskId = asyncTaskManager.runAsyncTask(
        new BatchUploadEndTask(actionId, tabletName, columnNames, mapFileIds, mapFilePaths));
    return taskId;
  }
  
  class BatchUploadEndTask extends AsyncTask {
    String actionId;
    String tabletName;
    String[] columnNames; 
    String[] mapFileIds;
    String[] mapFilePaths;
    
    public BatchUploadEndTask(String actionId,
        String tabletName, 
        String[] columnNames, 
        String[] mapFileIds,
        String[] mapFilePaths) {
      this.actionId = actionId;
      this.tabletName = tabletName;
      this.columnNames = columnNames;
      this.mapFileIds = mapFileIds;
      this.mapFilePaths = mapFilePaths;
      
      //LOG.debug("Start completing batch upload:" + tabletName + ",actionId=" + actionId);
    }
    
    public void exec() throws Exception {
      Map<String, BatchUploadAction> actions = null;
      synchronized(batchUploadActions) {
        actions = batchUploadActions.get(tabletName);
      }
      
      if(actions == null) {
        throw new IOException("No opened Uploader: " + tabletName);
      }
      
      synchronized(actions) {
        if(!actions.containsKey(actionId)) {
          throw new IOException("No opened Uploader: " + tabletName + "," + actionId + ",host=" + hostName);
        }
        Tablet tablet = getTablet(tabletName);
        if(tablet == null ) {
          throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
        }
        //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
        lockService.deleteNode(Constants.UPLOADER + "/" + tabletName + "/" + actionId, true);
        
        if(mapFileIds == null || mapFileIds.length == 0) {
          return;
        }      
        long indexSize = 0;
        List<TabletMapFile> mapFiles = new ArrayList<TabletMapFile>();
        try {
          for(int i = 0; i < mapFileIds.length; i++) {
            if(mapFileIds[i] == null) {
              continue;
            }
            TabletMapFile tabletMapFile = new TabletMapFile(conf,
                                                            NeptuneFileSystem.get(conf),
                                                            tablet.getTabletInfo(), 
                                                            columnNames[i], mapFileIds[i],
                                                            new GPath(mapFilePaths[i]),
                                                            tablet.getTable().getNumOfVersion());
            if(!fs.exists(new GPath(mapFilePaths[i], TabletMapFile.DATA_FILE)) ||
                !fs.exists(new GPath(mapFilePaths[i], TabletMapFile.IDX_FILE)))  {
              //Tablet을 재할달 받은 경우 File이 이미 이동 되었을 수도 있음.
              //FIXME 이전 TabletServer에서 data or index 파일 중 하나만 이동된 경우는?
              LOG.warn("End batch uploader: No data file or index file in " + mapFilePaths[i]);
              continue;
            }
            mapFiles.add(tabletMapFile);
            GPath targetPath = new GPath(Tablet.getTabletPath(conf, tablet.getTabletInfo()), 
                                              columnNames[i] + "/" + tabletMapFile.getFileId() + "/");
            tabletMapFile.moveTo(targetPath, true);
            tabletMapFile.loadIndex();
            indexSize += tabletMapFile.getIndexMemorySize();
            //LOG.debug("End batch uploader: add mapfile:" + tabletMapFile.getFileId());
            tablet.getDiskSSTable().addTabletMapFile(columnNames[i], tabletMapFile);
          }
        } catch (Exception e) {
          //map 파일 제거
          for(TabletMapFile eachMapFile: mapFiles) {
            try {
              tablet.getDiskSSTable().removeTabletMapFile(eachMapFile.getColumnName(), eachMapFile);
              eachMapFile.delete();
            } catch (Exception err) {
            }
          }
          LOG.error("endBatchUploader:" + e.getMessage(), e);
          throw new IOException(e.getMessage());
        } finally {
          LOG.debug("End endBatchUploader:" + actionId);
          tablet.actionChecker.endAction(actions.get(actionId));
          actions.remove(actionId);
          mapFiles.clear();
        }
      }
    }
  }
  
  public AsyncTaskStatus getAsyncTaskStatus(String taskId) throws IOException {
    return asyncTaskManager.getAsyncTaskStatus(taskId);
  }
  
  public void truncateColumn(String tabletName, String columnName) throws IOException {
    Tablet tablet = getTablet(tabletName);
    if(tablet == null) {
      throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
    }
    
    tablet.truncateColumn(columnName);
  }
  
  public void addColumn(String tableName, String tabletName, String addedColumnName) throws IOException {
    Tablet tablet = getTablet(tabletName);
    if(tablet == null) {
      throw new IOException("[" + tabletName + "] not serviced on this server(" + hostName + ")");
    }
    tablet.addColumn(addedColumnName);
  }
  
  public boolean isShutdowned() {
    return stopRequested;
  }

  /**
   * 
   * @author babokim
   * 
   */
  class TabletLoadTask implements Runnable {
    private TabletInfo tabletInfo;
    private TabletServer tabletServer;
    
    public TabletLoadTask(TabletServer tabletServer, TabletInfo tabletInfo) {
      this.tabletInfo = tabletInfo;
      this.tabletServer = tabletServer;
    }

    public void run() {
      try {
        if(!lockService.exist(tabletServerLockPath)) {
          LOG.error("Can't TabletServer lock while table load:" + tabletInfo);
          reportLoadError();
          return;
        }
      } catch (Exception e) {
        reportLoadError();
        LOG.error("Can't TabletServer lock while table load", e);
        return;
      }
      
      //Drop 처리 중인 Table인 경우
      String tableDropLockPath = Constants.TABLE_DROP + "/" + tabletInfo.getTableName();
      try {
        //if(LockServiceFactory.getLockService(conf, hostName, true).getMetaData(tableDropLockPath) != null) {
        if(lockService.getMetaData(tableDropLockPath) != null) {
          //Drop 처리 중인 Table인 경우
          NTable ntable = NTable.openTable(conf, TABLE_NAME_META);
          tablets.remove(tabletInfo.getTabletName());
          //META에서 Tablet 정보를 삭제한다.
          LOG.debug("Drop table:delete from META:" + tabletInfo);
          Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(), tabletInfo.getEndRowKey());
          ntable.remove(rowKey, Constants.META_COLUMN_NAME_TABLETINFO, new Cell.Key(tabletInfo.getTabletName()));
          return;
        }
      } catch (IOException e) {
        LOG.error(e.getMessage(), e);
      }
      
      LOG.info("TabletLoad start:" + tabletInfo);
      boolean create = false;
      Tablet tablet = null;
      try {
        create = !fs.exists(tabletInfo.getTabletPath());

        LOG.debug("tablet create : " + create);

        if (create) {
          tablet = createNewTablet();
        } else {
          tablet = loadExistedTablet();    
        }

        if(tablet == null) {
          LOG.error("Error while tablet loading: tablet is null: " + tabletInfo);
          return;
        }
        
        //BatchUploader가 수행되어 있는지 확인
        checkBatchUploader(tablet);
        
        //TabletServer의 서비스 목록에 추가
        lock.obtainWriteLock();
        try {
          tablet.setReady(true);
          LOG.info("tablet.put(" + tabletInfo.getTabletName() + ", " + tablet + ");");
          tablets.put(tabletInfo.getTabletName(), tablet);
        } finally {
          lock.releaseWriteLock();
        }        
      } catch (Exception e) {
        LOG.error("Tablet assign error tablet[" + tabletInfo.getTabletName() + "]", e);
        lock.obtainWriteLock();
        try {
          tablets.remove(tabletInfo.getTabletName());
        } finally {
          lock.releaseWriteLock();
        }

        reportLoadError();
        return;
      } 

      //LOG.info("End Tablet loading:" + tabletInfo);
      
      // Master로 end report
      try {
        proxyObject.endTabletAssignmentForHandler(tabletInfo, create);
      } catch (Exception e) {
        LOG.error("Error while reporting end of tablet load to master", e);
        // Report 중 에러가 발생한 경우 그냥 계속 서비스 한다.
      }
    }

    /**
     * @param e
     */
    private void reportLoadError() {
      //Master로 error report
      try {
        tabletInfo.setAssignedHostName(hostName);
        connectMasterServer().errorTabletAssignment(hostName, tabletInfo);
      } catch (IOException e1) {
        LOG.error("Tablet assign error: reportLoadError", e1);
      }
    }

    private void checkBatchUploader(Tablet tablet) throws IOException {
      //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
      Map<String, LockMetaData> actionIds = lockService.readDir(Constants.UPLOADER + "/" + tabletInfo.getTabletName());
      if(actionIds == null) {
        return;
      }
      for(LockMetaData lockMetaData: actionIds.values()) {
        String actionId = LockServiceFactory.getLockMetaDataPathName(lockMetaData); 
        if(lockService.getLock(lockMetaData.getLockId(), LockService.LockMode.LOCK_EXCLUSIVE)) {
          BatchUploadAction action = new BatchUploadAction(actionId);
          
          synchronized(batchUploadActions) {
            boolean result =  tablet.actionChecker.startAction(action);
            if(result) {
              Map<String, BatchUploadAction> actions = batchUploadActions.get(tabletInfo.getTabletName());
              if(actions == null) {
                actions = new HashMap<String, BatchUploadAction>();
                batchUploadActions.put(tabletInfo.getTabletName(), actions);
              }
              actions.put(action.getActionKey(), action);
            }
          }
        }
      }
    }
    
    private boolean lockRootTablet() throws IOException {
      LockMetaData lockMetaData = new LockMetaData(Constants.ROOT_TABLET);
      lockMetaData.setContentKey(tabletInfo.getTabletName() + "," + hostName);
      lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
      boolean success =lockService.createNode(lockMetaData, true);
      LOG.debug("Save ROOT INFO:" + success);
      return success;
//      GPath rootTabletInfoPath = new GPath(conf.get("neptune.root") + Constants.ROOT_TABLET );
//      GPath[] paths = fs.list(rootTabletInfoPath);
//      if(paths != null) {
//        for(GPath eachPath: paths) {
//          fs.delete(eachPath);
//        }
//      }
//      return fs.mkdirs(new GPath(rootTabletInfoPath, tabletInfo.getTabletName() + "," + NetworkUtil.hostNameToFileName(hostName)));
    }
    
    private Tablet createNewTablet() throws IOException {
      // create new tablet
      if(!fs.mkdirs(tabletInfo.getTabletPath())) {
        if(!fs.mkdirs(tabletInfo.getTabletPath())) {
          throw new IOException("Fail make tablet info on filesystem");
        }
      }
      
      tabletInfo.setAssignedHostName(hostName);
      //Tablet 생성
      Tablet tablet = null;
      try {
        TableSchema tableSchema = descTable(tabletInfo.getTableName());
        tablet = new Tablet(conf, tableSchema, tabletServer, tabletInfo);
      } catch(IOException e) {
        try {
          fs.delete(Tablet.getTabletPath(conf, tabletInfo), true);
        } catch (Exception e1) {}
        LOG.error("Error while create new Tablet:" + tabletInfo, e);
        throw e;
      }
      
      // insert tablet info to META or lock ROOT
      if (TABLE_NAME_ROOT.equals(tabletInfo.getTableName())) {
        if(!lockRootTablet()) {
          LOG.warn("ROOT Tablet already serviced by other TabletServer");
          //이미 다른 Tablet에서 서비스 하고 있는 경우
          return null;
        }
      } else {
        try {
          if (TABLE_NAME_META.equals(tabletInfo.getTableName())) {
            insertIntoRootTable(tabletInfo, System.currentTimeMillis());
          } else {
            insertIntoMetaTable(tabletInfo, System.currentTimeMillis());
          }
        } catch (IOException e) {
          //오류 발생시 생성된 파일 삭제
          try {
            fs.delete(Tablet.getTabletPath(conf, tabletInfo), true);
          } catch (Exception e1) {}
          throw e;
        }
      }
      
      return tablet;
    }

    private Tablet loadExistedTablet() throws IOException {
      // load existed tablet
      LOG.debug("Start loadExistedTable: " + tabletInfo);
      
      tabletInfo.setAssignedHostName(hostName);
      Tablet tablet = getTablet(tabletInfo.getTabletName());
      if(tablet != null) {
        //이미 서비스 하고 있는 경우
        LOG.debug("The tablet is already served. return");
        return null;
      }
      
      TableSchema tableSchema = descTable(tabletInfo.getTableName());
      if(tableSchema == null) {
        LOG.error("No Table Schema:" + tabletInfo.getTableName());
        throw new IOException("No Table Schema:" + tabletInfo.getTableName());
      }
      tablet = new Tablet(conf, tableSchema, tabletServer, tabletInfo);
      if(!tablet.load()) {
        LOG.debug("Call minor compaction cause fail loading change log");
        tablet.doMinorCompaction();
      }
      
      //delete drop lock
      String lockPath = Constants.TABLET_DROP + "/" + 
                          tabletInfo.getTableName() + "/" + tabletInfo.getTabletName();
      
      //LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
      if(lockService.getMetaData(lockPath) != null && !lockService.deleteNode(lockPath, true)) {
        throw new IOException("Can't delete tablet drop lock:" + lockPath);
      }
      
      // modify tablet service hostname in metadata
      if (TABLE_NAME_ROOT.equals(tabletInfo.getTableName())) {
        if(!lockRootTablet()) {
          //lock을 가져 올수 없으면 다른 TabletServer에서 서비스하고 있는 경우
          LOG.warn("ROOT Tablet already serviced by other TabletServer");
          return null;
        }
      } else if (TABLE_NAME_META.equals(tabletInfo.getTableName())) {
        modifyRootTable(tabletInfo);
      } else {
        //LOG.debug("Start modifyMetaTable: " + tabletInfo);
        modifyMetaTable(tabletInfo);
      }
      
      LOG.debug("End loadExistedTable: " + tabletInfo);
      return tablet;
    }
  } // end of TabletLoadThread class

  class MinorCompactionThread extends Thread {
    private long maxMemorySSTableSize;
    private long forceCompactionMemory;
    private long maxMemoryPerTablet;
    private long maxMemory;
    public MinorCompactionThread() {
      super("MinorCompactionThread");
      maxMemoryPerTablet = conf.getLong("memory.maxTabletSizePerTablet", 30) * 1024L * 1024L;
      maxMemorySSTableSize = conf.getLong("memory.maxTabletSize", 400) * 1024L * 1024L;
      maxMemory = SizeOf.maxMemory();

      float maxFreeMemoryRate = conf.getFloat("memory.maxFreeMemoryRate", 0.2f);
      forceCompactionMemory = (long)((float)maxMemory * maxFreeMemoryRate);
    }

    public void run() {
      long elapseTime = 0;
      long startTime = System.currentTimeMillis();
      
      while (!stopRequested) {
        elapseTime = System.currentTimeMillis() - startTime;
        if(elapseTime < 5 * 1000) {
          try {
            Thread.sleep(5 * 1000);
          } catch (InterruptedException e) {
            return;
          }
        }
        startTime = System.currentTimeMillis();
        
        try {
          if(lockService == null) {
            lockService = LockServiceFactory.getLockService(conf, hostName, true, heartbeatPort);
            continue;
          }
          
          if(!lockService.exist(tabletServerLockPath)) {
            LOG.error("Can't TabletServer lock while MinorCompactionThread");
            continue;
          }

          
          long freeMemory = SizeOf.freeMemory();
          long totalMemory = SizeOf.totalMemory();
          
          //TabletServer의 메모리가 부족한 경우 모든 Tablet에 대해 minor compaction을 수행한다.
          if((maxMemory - (totalMemory - freeMemory)) < forceCompactionMemory) {
            SizeOf.runGC();
            freeMemory = SizeOf.freeMemory();
            totalMemory = SizeOf.totalMemory();
            if((maxMemory - (totalMemory - freeMemory)) < forceCompactionMemory) {
              LOG.warn("Not enough free memory:freeMemory=" + freeMemory + ", maxMemory=" + maxMemory + ", all tablets do minor compaction");
              LOG.info("Set HeavyMemory true.");
              heavyMemory.getAndSet(true);
              forceMinorCompactionAllTablets();
              continue;
            } else {
              if(heavyMemory.get()) {
                LOG.info("Set HeavyMemory reset to false.");
              }
              heavyMemory.getAndSet(false);
            }
          } else {
            if(heavyMemory.get()) {
              LOG.info("Set HeavyMemory reset to false.");
            }
            heavyMemory.getAndSet(false);
          }
          
          //minor compaction 대상이 되는 tablet과 전체 Tablet의 MemorySSTable 사이즈를 가져온다.
          List<Tablet> compactionTargets = new ArrayList<Tablet>();
          int sumSize = 0;
          lock.obtainReadLock();
          try {
            for (Tablet tablet : tablets.values()) {
              int memTableSize = tablet.getMemtableSize();
              
              if(memTableSize >= maxMemoryPerTablet) {
                compactionTargets.add(tablet);
              }
              sumSize += memTableSize;
            }
          } finally {
            lock.releaseReadLock();
          }
  
          //MemorySSTable의 전체 크기가 정해진 사이즈보다 큰 경우 
          //메모리를 많이 차지하는 Top-N에 대해 minorCompaction 수행
          if (sumSize > maxMemorySSTableSize) {
            //LOG.debug("MinorCompactionThread: start doMinorCompaction: sumSize : " + sumSize + ", maxSize : " + maxMemorySSTableSize);
            // Tablet중 메모리 사이즈가 가장 큰 Tablet을 compaction한다.
            Tablet[] targetTablets = findMaxMemoryTablet();
            if(targetTablets == null || targetTablets.length == 0) {
              LOG.warn("MemorySSTable is full. but all tablets ready stop");
              continue;
            }
            for(Tablet tablet: targetTablets) {
              if(Constants.TABLE_NAME_META.equals(tablet.getTabletInfo().getTableName()) || 
                  Constants.TABLE_NAME_ROOT.equals(tablet.getTabletInfo().getTableName())) {
                continue;
              }
              
              try {
                tablet.doMinorCompaction();
                Thread.sleep(200);
              } catch (Exception e) {
                LOG.error("MinorCompaction Error2:" + tablet.getTabletInfo() + "," + e.getMessage(), e);
              }
            }
            //LOG.debug("MinorCompactionThread: end doMinorCompaction:" + sumSize);
          } else if(compactionTargets.size() > 0) {
            //memory tablet size를 초과한 Tablet에 대해 compation을 수행한다.
            //LOG.debug("MinorCompactionThread: start doMinorCompaction:" + compactionTargets.size());
            for(Tablet eachTablet: compactionTargets) {
              eachTablet.doMinorCompaction();
              Thread.sleep(200);
            }
            //LOG.debug("MinorCompactionThread: end doMinorCompaction");
          }
        } catch (Exception e) {
          LOG.error("MinorCompactionThread Error:" + e.getMessage(), e);
        }
      }
    }

    private void forceMinorCompactionAllTablets() {
      List<Tablet> tabletList = new ArrayList<Tablet>();
      lock.obtainReadLock();
      try {
        tabletList.addAll(tablets.values());
      } finally {
        lock.releaseReadLock();
      }
      for(Tablet eachTablet: tabletList) {
        if(Constants.TABLE_NAME_META.equals(eachTablet.getTabletInfo().getTableName()) || 
            Constants.TABLE_NAME_ROOT.equals(eachTablet.getTabletInfo().getTableName())) {
          continue;
        }
        try {
          //현재 수행중인 Action  정보를 로그로 남긴다.
          String[] actions = eachTablet.getAllActions();
          if(actions != null) {
            boolean printed = false;
            for(String eachAction: actions) {
              if(eachAction.indexOf("Running Actions") >= 0 ||
                  eachAction.indexOf("Waiting Actions") >= 0) {
                continue;
              }
              printed = true;
              if(eachAction.indexOf("ApplyAction") >= 0) {
                continue;
              }
              LOG.debug(eachTablet.getTabletInfo() + ": Action:" + eachAction);
            }
            if(printed) {
              LOG.debug(eachTablet.getTabletInfo() + " memory=" + eachTablet.getMemtableSize());
            }
          }
          
          //minor compaction을 수행한다.
          eachTablet.doMinorCompaction();
          Thread.sleep(200);
        } catch (Exception e) {
          LOG.error("MinorCompaction Error:" + eachTablet.getTabletInfo() + "," + e.getMessage(), e);              
        }
      }
    }

    public Tablet[] findMaxMemoryTablet() {
      List<Tablet> tabletList = new ArrayList<Tablet>();
      lock.obtainReadLock();
      try {
        tabletList.addAll(tablets.values());
      } finally {
        lock.releaseReadLock();
      }
      Collections.sort(tabletList, new Comparator<Tablet>() {
        public int compare(Tablet tablet1, Tablet tablet2) {
          int size1 = tablet1.getMemtableSize();
          int size2 = tablet2.getMemtableSize();
          if (size1 == size2) {
            return 0;
          } else if (size1 > size2) {
            return 1;
          } else {
            return -1;
          }
        }
      });

      int MAX_CONCURRENT_MINOR_SIZE = 20;
      if(tabletList.size() < MAX_CONCURRENT_MINOR_SIZE) {
        return tabletList.toArray(new Tablet[]{});
      } else {
        Tablet[] targetTablets = new Tablet[MAX_CONCURRENT_MINOR_SIZE];
        for(int i = 1; i <= MAX_CONCURRENT_MINOR_SIZE; i++) {
          targetTablets[i -1] = tabletList.get(tabletList.size() - i);  
        }
        return targetTablets;  
      }
    }
  } // end of MinorCompactionThread class

  class MajorCompactionThread extends Thread {
    public MajorCompactionThread() {
      super("MajorCompactionThread");
    }
    
    public void run() {
      Map<String, Tablet> tempTablets = new HashMap<String, Tablet>(10);
      long elapseTime = 0;
      long startTime = System.currentTimeMillis();
      while (!stopRequested) {
        tempTablets.clear();
        
        elapseTime = System.currentTimeMillis() - startTime;
        if(elapseTime < 5 * 1000) {
          try {
            Thread.sleep(5 * 1000);
          } catch (InterruptedException e) {
            return;
          }
        }
        startTime = System.currentTimeMillis();
        
        if(compactionExecutor.getActiveCount() > maxMajorCompactionThread / 2 ||
            splitExecutor.getActiveCount() > maxSplitThread / 2) {
          heavyThread.getAndSet(true);
          LOG.debug("Heavy Thread:" + compactionExecutor.getActiveCount() + "," + splitExecutor.getActiveCount());
        } else {
          if(heavyThread.get()) {
            LOG.debug("Clear Heavy Thread");
          }
          heavyThread.getAndSet(false);
        }
        
        try {
          if(lockService == null) {
            lockService = LockServiceFactory.getLockService(conf, hostName, true, heartbeatPort);
            continue;
          }

          if(!lockService.exist(tabletServerLockPath)) {
            LOG.error("Can't TabletServer lock while MajorCompactionThread");
            continue;
          }
        } catch (IOException e1) {
          LOG.error("Can't TabletServer lock while MajorCompactionThread", e1);
          continue;
        }
        
        synchronized (tablets) {
          tempTablets.putAll(tablets);
        }
        
        int splitCount = 0;
        for (Map.Entry<String, Tablet> entry : tempTablets.entrySet()) {
          try {
            Tablet tablet = entry.getValue();
            
            if (!tablet.isReady()) {
              continue;
            }
//            if(Constants.TABLE_NAME_META.equals(tablet.getTabletInfo().getTableName()) || 
//                Constants.TABLE_NAME_ROOT.equals(tablet.getTabletInfo().getTableName())) {
//              continue;
//            }
            //MajorCompaction
            if(tablet.neededMajorCompaction()) {
              tablet.doMajorCompaction();
            }
            
            //split이 처리되지 않는 경우 rate를 낮게 적용해서 2개 씩 split을 유도한다.
            double rate = 0.9;
            if(splitExecutor.getActiveCount() < 2 && splitCount < 2) {
              rate = 0.7;
            }
            if(tablet.neededSplit(rate)) {
              tablet.doTabletSplit();
              splitCount++;
            }
          } catch (IOException e) {
            LOG.error("MajorCompactionThread error: " + e.getMessage(), e);
          }
        }
      }
    }
  } // end of MajorCompatcionThread class

  class TxListener implements LeaseListener {
    String tabletName;

    TxId txId;

    public TxListener(String tabletName, TxId txId) {
      this.tabletName = tabletName;
      this.txId = txId;
    }

    public void expire() {
      try {
        Row.Key rowKey = rollback(txId);
        LOG.info("Tx expired:txId=" + txId + ", rowKey=" + rowKey);
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  } // end of TxListener class

  class TabletScannerListener implements LeaseListener {
    String tabletName;

    String scannerId;

    public TabletScannerListener(String tabletName, String scannerId) {
      this.tabletName = tabletName;
      this.scannerId = scannerId;
    }

    public void expire() {
      try {
        LOG.info("Scanner expired:" + scannerId);
        closeScanner(scannerId);
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  } // end of TabletScannerListener class

  /**
   * Scanner를 위한 서버
   * @author babokim
   *
   */
  class DataXceiveServer implements Runnable {
      boolean shouldListen = true;
      ServerSocket ss;
      public DataXceiveServer(ServerSocket ss) {
          this.ss = ss;
      }

      /**
       */
      public void run() {
          try {
              while (shouldListen) {
                  Socket s = ss.accept();
                  new Daemon(new DataXceiver(s)).start();
              }
              ss.close();
          } catch (IOException ie) {
              LOG.info("Exiting ScannerXceiveServer due to " + ie.toString());
          }
      }
      
      public void kill() {
          this.shouldListen = false;
          try {
              this.ss.close();
          } catch (IOException iex) {
          }
      }
  }

  class TabletServerLockChecker implements Runnable {
    private boolean sendReportToMaster = false;
    public void run() {
      while(true) {
        try {
          Thread.sleep(10 * 1000);
        } catch (InterruptedException e) {
          return;
        }    
        
        //NeptuneMaster로 자신이 startup 되었는지 여부를 전송한다.
        if(!sendReportToMaster) {
          try {
            TabletMasterProtocol master = connectMasterServer();
            if (master == null) {
              LOG.warn("Cannot connect to NeptuneMaster. Sending report is ignored");
              continue;
            }
            
            master.reportTabletServerStarted(hostName);
            sendReportToMaster = true;
          } catch (IOException e) {
            LOG.warn("Can't send reportTabletServerStarted" ,e);
            sendReportToMaster = false;
          }
        }
        
        //lock 파일이 존재하지 않으면 종료
        try {
          checkTabletServerLock();
        } catch(RuntimeException e) { // RPCInterruptedException of neptune or pleiades
          LOG.debug("TabletServerLockChecker is gracefully closed");
          return;
        }
      }      
    }
  }
  
  private static class Count {
    int value = 0;
    Count(int init) { value = init; }
    synchronized void incr() { value++; }
    synchronized void decr() { value--; }
    synchronized int getValue() { return value; }
    public String toString() { return Integer.toString(value); }
    
  }
  /**
   * Thread for processing incoming/outgoing data stream
   */
  class DataXceiver implements Runnable {
    Socket socket;
    DataOutputStream out;
    DataInputStream in;
    byte TEST_OP = (byte) 61;
    Row.Key rowKey = new Row.Key("00000109308");
    String[] columns = new String[]{"Col1"};
    
    public DataXceiver(Socket socket) {
        this.socket = socket;
        //LOG.debug("Number of active connections is: "+ activeDataDaemons.incrementAndGet());
    }

    public void run() {
//      if(scanCount.getValue() > 20) {
//        //LOG.warn("Too many opened scanner: " + scanCount + ".");
//        try {
//          socket.close();
//        } catch (IOException e) {
//        }
//        return;
//      }
      
      try {
        out = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
        in = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
        
        byte op = (byte)in.read();
        if(op == OPEN_SCANNER) {
          runScan();
        } else if(op == TEST_OP) {
          while(true) {
            op = (byte)in.read();
            if(op != (byte) 62) {
              break;
            }
            ColumnValue[][] result = get("M_DATA_1000000_385860558163441", rowKey, columns);
            //WritableUtils.writeString(out, "AAA");
            result[0][0].write(out);
            out.flush();
          }
        } else {
          LOG.error("DataXceiver wrong op code:" + op);
        }
      } catch (Exception e) {
        LOG.error("DataXceiver run error: " + e.getMessage(), e);        
      } finally {
        //activeDataDaemons.decrementAndGet();
        if(out != null) { try { out.close(); } catch (Exception e) {} }
        if(in != null) { try { in.close(); } catch (Exception e) {} }
        try {
          socket.close();
        } catch (Exception e) {}
        
      }
    }
    
    private void runScan() throws Exception {
      //FIXME Scanner open시 timeout이 걸렸을 떄 ScannerAction을 삭제시켜주는 로직
      scanCount.incr();
      
      //LOG.debug(hostName + " Number of active sacnner daemon: " + scanCount);
      
      String scannerId = null;
      String tabletName = null;
      try {
        tabletName = NWritableUtils.readString(in);
        //LOG.debug(hostName + " Number of active sacnner daemon: " + scanCount + ",tabletName=" + tabletName);
        String columnName = NWritableUtils.readString(in);
        RowFilter scanFilter = new RowFilter();
        scanFilter.readFields(in);
        
        //LOG.debug(hostName + " Number of active sacnner daemon: " + scanCount + ",tabletName=" + tabletName);
        int retry = 0;
        int maxRetry = 10;
        try {
          tablets.get(tabletName);
          while(retry < maxRetry) {
            scannerId = openScanner(tabletName, columnName, scanFilter);
            if(scannerId != null) {
              break;
            }
            retry++;
            try {
              Thread.sleep(1 * 1000);
            } catch (InterruptedException e) {
              return;
            }
          }
          if(retry >= maxRetry && scannerId == null) {
            LOG.error("Scanner open timeout");
            out.write(SCANNER_OPEN_FAIL);
            NWritableUtils.writeString(out, "Scanner open timeout");
            out.flush();
            return;
          }
        } catch (Exception e) {
          LOG.error("Scanner open error:" + e.getMessage());
          out.write(SCANNER_OPEN_FAIL);
          NWritableUtils.writeString(out, e.getMessage());
          out.flush();
          return;
        }
        
        out.write(SCANNER_OPEN_SUCCESS);
        NWritableUtils.writeString(out, scannerId);
        out.flush();

        //FIXME 데이터 전송 중에 에러 발생시 클라이언트에서 이를 어떻게 식별할 것인가?
        long nextCount = 0;
        if(scannerId != null) {
          ColumnValue[] columnValues = null;
          while( (columnValues = next(scannerId))  != null ) {
            out.writeInt(columnValues.length);
            out.flush();
            for(int i = 0; i < columnValues.length; i++) {
              columnValues[i].write(out);
            }
            out.flush();
            nextCount++;
            if(nextCount % 10000 == 0) {
              Thread.sleep(50);
            }
            
            if(nextCount % 10000 == 0) {
              LOG.debug(tabletName + " " + nextCount + " scaned");
            }
          }
          out.writeInt(-1);
          out.flush();
        }
      } catch ( SocketException ignored ) {
        
      } catch(Exception e) {
        LOG.error("runScan", e);
      } finally {
        scanCount.decr();
        if(scannerId != null) {
          try {
            closeScanner(scannerId);
          } catch (Exception e) {
            LOG.error(e);
          }
        }
      }
    }
  }
  
  public String test(long sleepTime, String echo) throws IOException {
    if(sleepTime > 0) {
      try {
        Thread.sleep(sleepTime);
      } catch (InterruptedException e) {
        return null;
      }
    }
    return echo;
  }
  
  Object getMonitor = new Object();
  class GetThread extends Thread {
    int threadNum = 0;
    public GetThread(int threadNum) {
      this.threadNum = threadNum;
    }
    public void run() {
      List<GetExecThread> threads = new ArrayList<GetExecThread>();
      long startTime = System.currentTimeMillis();
      //LOG.info("threadNum:" + threadNum);
      for(int i = 0; i < threadNum; i++) {
        GetExecThread thread = new GetExecThread();
        thread.start();
        threads.add(thread);
      }
      int endCount = 0;
      
      while(true) {
        synchronized(getMonitor) {
          try {
            getMonitor.wait();
          } catch (InterruptedException e) {
          }
          endCount++;
          if(endCount == threadNum) {
            break;
          }
        }
      }
      long endTime = System.currentTimeMillis();
      
      LOG.info("TotalTime:" + (endTime - startTime) + "," + (threadNum * 100000));
    } 
  }
  
  class GetExecThread extends Thread {
    int count = 0;
    public void run() {
      try {
        Row.Key rowKey = new Row.Key("00000109308");
        String[] columns = new String[]{"Col1"};
        long startTime = System.currentTimeMillis();
        LOG.info("Start get");
        GPath path = new GPath("/user/neptune/table/M_INPUT_1000000/M_INPUT_1000000_622318953150539/Col1/622314891337687/map.dat");
        while(true) {
//          ColumnValue[][] result = get("M_DATA_1000000_385860558163441", rowKey, columns);
//          if(result == null || result.length == 0 || result[0] == null) {
//            LOG.info("Result is null. test stop");
//            break;
//          }
          DataInputStream in = fs.openDataInputStream(path);
          byte[] buf = new byte[1024];
          in.read(buf);
          in.close();
          count++;
          if(count > 100000) {
            break;
          }
        }
        long endTime = System.currentTimeMillis();
        LOG.info("Thread Time:" + (endTime - startTime) + "," + count);
      } catch (Exception e) {
        LOG.error(e);
      }
      synchronized(getMonitor) {
        getMonitor.notifyAll();
      }
    }
  }
  
  public void printMeta(String tabletName) throws IOException {
    Tablet tablet = tablets.get(tabletName);
    tablet.print();
  }
  
  public TabletReport getTabletDetailInfo(String tabletName) throws IOException {
    Tablet tablet = tablets.get(tabletName);
    if(tablet != null) {
      return tablet.getTabletDetailInfo();
    } else {
      return null;
    }
  }
  
  public boolean canUpload() throws IOException {
    if(uploaderCount.value > 2) {
      return false;
    } else {
      return true;
    }
  }
  
  public void doActionForTest(String tabletName, String action) throws IOException {
    final Tablet tablet = tablets.get(tabletName);
    if(tablet == null) {
      throw new IOException("No Tablet:" + tabletName + ", host=" + hostName);
    }
    if("minorCompaction".equals(action)) {
      tablet.doMinorCompaction();
    } else if("majorCompaction".equals(action)) {
      tablet.doMajorCompaction();
    } else if("split".equals(action)) {
      Thread t = new Thread() {
        public void run() {try {
            tablet.doMinorCompaction();
            while(true) {
              if(tablet.diskSSTable.getMaxMapFileCount() == 1) {
                break;
              } else if(tablet.diskSSTable.getMaxMapFileCount() > 1) {
                tablet.doMajorCompaction();
              } else {
                try {
                  Thread.sleep(1000);
                } catch (InterruptedException e) {
                  return;
                }          
              }
            }
            tablet.doTabletSplit();
        } catch (Exception e) {
          e.printStackTrace(System.out);
        }
        }
      };
      t.start();
    } else {
      throw new IOException("Invalid action:" + action);
    }
  }
  
  public String getTestHandlerKey() {
    return hostName;
  }
  
  public ColumnValue[][] testRPC() {
    ColumnValue[][] result = new ColumnValue[2][];
    
    return result;
  }
  
  public void setProperty(String key, String value) throws IOException {
    conf.set(key, value);
  }

  public String getProperty(String key, String defaultValue) throws IOException {
    return conf.get(key, defaultValue);
  }
  
  public ColumnValue[] getAllMemoryValues(String tabletName, String columnName) throws IOException {
    Tablet tablet = getTablet(tabletName);
    if(tablet == null) {
      return null;
    }
    
    return tablet.getMemorySSTable().getAllMemoryValues(columnName);
  }

  public NConfiguration getConf() {
    return conf;
  }
  
  public String getTabletServerLockPath() {
    return tabletServerLockPath;
  }
  
  public TabletServerStatus getServerStatus() throws IOException {
    TabletServerStatus status = new TabletServerStatus();
    
//    List<Tablet> copyedTablets = new ArrayList<Tablet>();
//    copyedTablets.addAll(tablets.values());
//    
//    TabletReport[] reports = new TabletReport[copyedTablets.size()];
//    int index = 0;
//    for(Tablet eachTablet: copyedTablets) {
//      reports[index] = eachTablet.getTabletDetailInfo();
//    }
//    
//    status.setTabletReports(reports);
    
    status.setServerStartTime(serverStartTime.getTime());
    status.setFreeMemory(Runtime.getRuntime().freeMemory());
    status.setMaxMemory(Runtime.getRuntime().maxMemory());
    status.setTotalMemory(Runtime.getRuntime().totalMemory());

    status.setNumMajorThread(compactionExecutor.getActiveCount());
    status.setNumSplitThread(splitExecutor.getActiveCount());
    status.setNumEtcThread(actionExecutor.getActiveCount());
    
    status.setTxStatistics(txStatistics.getTxStatistics());
    return status;
  }
  
  
  public boolean hasValue(String tabletName, String columnName, Row.Key rowKey) throws IOException {
    //TimeLog.checkPoint();
    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }
    return tablet.hasValue(columnName, rowKey);
  }  
  
  @Override
  public Row.Key[] getSplitedRowKeyRanges(String tabletName, int splitPerTablet) throws IOException {
    Tablet tablet = getTablet(tabletName);
    if (tablet == null) {
      throw new IOException("no tablet[" + tabletName + "]");
    }
    return tablet.getSplitedRowKeyRanges(splitPerTablet);
  }
}
